mirror of
https://github.com/pgsty/minio.git
synced 2026-11-09 17:21:51 +03:00
Integrate R2 restored-copy cleanup safety with R1
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
@@ -0,0 +1,284 @@
|
||||
// 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"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||
"github.com/minio/minio/internal/bucket/versioning"
|
||||
xhttp "github.com/minio/minio/internal/http"
|
||||
)
|
||||
|
||||
const restoreSafetyRules = `<LifecycleConfiguration><Rule><ID>tier</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter><Transition><Days>1000</Days><StorageClass>WARM</StorageClass></Transition></Rule></LifecycleConfiguration>`
|
||||
|
||||
// Reuse the original restore-copy fixture with a small, isolated real cluster.
|
||||
func newRestoreSafetyTest(t *testing.T, pools int, state versioning.State) *restoreTest {
|
||||
t.Helper()
|
||||
prevTier := globalTierConfigMgr
|
||||
t.Cleanup(func() { globalTierConfigMgr = prevTier })
|
||||
dirs, err := getRandomDisks(4 * pools)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { removeRoots(dirs) })
|
||||
var endpoints EndpointServerPools
|
||||
for i := range pools {
|
||||
endpoints = append(endpoints, mustGetPoolEndpoints(i, dirs[4*i:4*i+4]...)...)
|
||||
}
|
||||
obj, _, err := initObjectLayer(t.Context(), endpoints)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
z := obj.(*erasureServerPools)
|
||||
t.Cleanup(func() { z.Shutdown(context.Background()) })
|
||||
globalTierConfigMgr = NewTierConfigMgr()
|
||||
globalTierConfigMgr.drivercache["WARM"] = &restoreTestTier{tierExpiryTestBackend: &tierExpiryTestBackend{objects: map[string][]byte{}}}
|
||||
r := &restoreTest{t: t, z: z, bucket: getRandomBucketName(), object: "obj", labels: map[string]string{}, bodies: map[string][]byte{}}
|
||||
if err := z.MakeBucket(t.Context(), r.bucket, MakeBucketOptions{VersioningEnabled: state != ""}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
meta, err := globalBucketMetadataSys.Get(r.bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if state == versioning.Suspended {
|
||||
meta.VersioningConfigXML = []byte(`<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>`)
|
||||
}
|
||||
meta.LifecycleConfigXML = []byte(restoreSafetyRules)
|
||||
if err := meta.Save(t.Context(), z); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
globalBucketMetadataSys.Set(r.bucket, meta)
|
||||
return r
|
||||
}
|
||||
|
||||
type restoreCleanupFaultDisk struct {
|
||||
consistencyReadFaultDisk
|
||||
deletes *atomic.Int64
|
||||
}
|
||||
|
||||
func (d restoreCleanupFaultDisk) DeleteVersion(ctx context.Context, bucket, object string, fi FileInfo, force bool, opts DeleteOptions) error {
|
||||
if bucket == d.bucket && object == d.object {
|
||||
d.deletes.Add(1)
|
||||
}
|
||||
return d.StorageAPI.DeleteVersion(ctx, bucket, object, fi, force, opts)
|
||||
}
|
||||
|
||||
func TestExpireRestoredRequiresFreshMetadata(t *testing.T) {
|
||||
for _, restored := range []bool{true, false} {
|
||||
t.Run(fmt.Sprintf("expire-restored=%v", restored), func(t *testing.T) {
|
||||
r := newRestoreSafetyTest(t, 1, "")
|
||||
oi := r.put("v1", true, true)
|
||||
r.restore(oi)
|
||||
before := r.fileInfo(oi.VersionID)
|
||||
er := r.z.serverPools[0].getHashedSet(r.object)
|
||||
original := er.getDisks
|
||||
disks := original()
|
||||
var deletes atomic.Int64
|
||||
faulty := make([]StorageAPI, len(disks))
|
||||
for i, d := range disks {
|
||||
faulty[i] = restoreCleanupFaultDisk{consistencyReadFaultDisk{d, r.bucket, r.object}, &deletes}
|
||||
}
|
||||
er.getDisks = func() []StorageAPI { return faulty }
|
||||
// Restore the callback before any fixture reads and on a fatal assertion.
|
||||
defer func() { er.getDisks = original }()
|
||||
opts := ObjectOptions{VersionID: nullVersionID, Transition: TransitionOptions{ExpireRestored: restored}, Expiration: ExpirationOptions{Expire: restored}}
|
||||
_, err := er.DeleteObject(t.Context(), r.bucket, r.object, opts)
|
||||
er.getDisks = original
|
||||
if restored {
|
||||
var quorum InsufficientWriteQuorum
|
||||
if !errors.As(err, &quorum) {
|
||||
t.Errorf("cleanup without fresh metadata: got %v, want write-quorum error", err)
|
||||
}
|
||||
if n := deletes.Load(); n != 0 {
|
||||
t.Errorf("cleanup attempted %d blind disk deletes", n)
|
||||
}
|
||||
if after := r.fileInfo(oi.VersionID); !reflect.DeepEqual(before, after) {
|
||||
t.Error("failed metadata read changed the stored version")
|
||||
}
|
||||
if n := r.dataDirs(before.DataDir); n != len(disks) {
|
||||
t.Errorf("cleanup removed data: %d/%d directories remain", n, len(disks))
|
||||
}
|
||||
} else {
|
||||
// Ordinary unversioned DELETE retains the existing tryDel behavior.
|
||||
if err != nil || deletes.Load() != int64(len(disks)) {
|
||||
t.Fatalf("ordinary DELETE: err=%v, deletes=%d", err, deletes.Load())
|
||||
}
|
||||
if _, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, ObjectOptions{}); !isErrObjectNotFound(err) {
|
||||
t.Fatalf("ordinary DELETE left the object: %v", err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestExpireRestoredRenewedOrMissing(t *testing.T) {
|
||||
r := newRestoreSafetyTest(t, 1, versioning.Enabled)
|
||||
oi := r.put("v1", false, true)
|
||||
r.restore(oi)
|
||||
stale, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(oi.VersionID))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := r.fileInfo(oi.VersionID)
|
||||
renewed := completedRestoreObj(time.Now().Add(48 * time.Hour)).String()
|
||||
opts := r.opts(oi.VersionID)
|
||||
opts.EvalMetadataFn = func(o *ObjectInfo, _ error) (ReplicateDecision, error) {
|
||||
o.UserDefined[xhttp.AmzRestore] = renewed
|
||||
return ReplicateDecision{}, nil
|
||||
}
|
||||
if _, err := r.z.PutObjectMetadata(t.Context(), r.bucket, r.object, opts); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = expireTransitionedObject(t.Context(), r.z, &stale, lifecycle.Event{Action: lifecycle.DeleteRestoredAction}, lcEventSrc_Scanner)
|
||||
if !isErrVersionNotFound(err) {
|
||||
t.Errorf("stale task should be rejected, got %v", err)
|
||||
}
|
||||
if got := r.fileInfo(oi.VersionID); got.Metadata[xhttp.AmzRestore] != renewed || got.DataDir != before.DataDir {
|
||||
t.Error("stale task changed renewed restore")
|
||||
}
|
||||
if n := r.dataDirs(before.DataDir); n != 4 {
|
||||
t.Errorf("renewed copy has %d directories, want 4", n)
|
||||
}
|
||||
if _, data, err := r.read(oi.VersionID); err != nil || !bytes.Equal(data, r.bodies["v1"]) {
|
||||
t.Fatalf("renewed copy not readable: %v", err)
|
||||
}
|
||||
|
||||
for _, id := range []string{mustGetUUID(), nullVersionID} {
|
||||
opts := r.opts(id)
|
||||
opts.Expiration.Expire = true
|
||||
opts.Transition.ExpireRestored = true
|
||||
if _, err := r.z.DeleteObject(t.Context(), r.bucket, r.object, opts); !isErrVersionNotFound(err) {
|
||||
t.Errorf("missing version %s: %v", id, err)
|
||||
}
|
||||
}
|
||||
missing := r.opts(nullVersionID)
|
||||
missing.Expiration.Expire = true
|
||||
missing.Transition.ExpireRestored = true
|
||||
if _, err := r.z.DeleteObject(t.Context(), r.bucket, "missing-object", missing); !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
|
||||
t.Errorf("missing object: %v", err)
|
||||
}
|
||||
if got := r.fileInfo(oi.VersionID); got.Metadata[xhttp.AmzRestore] != renewed {
|
||||
t.Error("missing-target cleanup changed existing restore")
|
||||
}
|
||||
}
|
||||
|
||||
func TestExpireRestoredStorageBoundary(t *testing.T) {
|
||||
for _, status := range []string{"", lifecycle.TransitionPending, lifecycle.TransitionComplete} {
|
||||
for _, inline := range []bool{false, true} {
|
||||
for _, requestStatus := range []string{"", lifecycle.TransitionComplete} {
|
||||
t.Run(fmt.Sprintf("stored=%s/inline=%v/request=%s", status, inline, requestStatus), func(t *testing.T) {
|
||||
disk, _, err := newXLStorageTestSetup(t)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
const bucket, object = "restore-storage", "obj"
|
||||
ctx := t.Context()
|
||||
if err := disk.MakeVol(ctx, bucket); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fi := newFileInfo(object, 2, 2)
|
||||
fi.VersionID, fi.DataDir = mustGetUUID(), mustGetUUID()
|
||||
fi.ModTime, fi.Size = time.Now().UTC(), 1024
|
||||
fi.TransitionStatus = status
|
||||
fi.AddObjectPart(1, "", fi.Size, fi.Size, fi.ModTime, nil, nil)
|
||||
fi.Metadata = map[string]string{xhttp.AmzRestore: completedRestoreObj(time.Now().Add(-time.Hour)).String(), "x-amz-meta-preserve": "keep"}
|
||||
payload := []byte("sole local contents")
|
||||
if inline {
|
||||
fi.Data = payload
|
||||
fi.SetInlineData()
|
||||
}
|
||||
if err := disk.WriteMetadata(ctx, "", bucket, object, fi); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
dir := filepath.Join(disk.Endpoint().Path, bucket, object, fi.DataDir)
|
||||
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(dir, "part.1"), payload, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, err := disk.ReadXL(ctx, bucket, object, true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var xl xlMetaV2
|
||||
if err := xl.Load(raw.Buf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
del := FileInfo{VersionID: fi.VersionID, ExpireRestored: true, TransitionStatus: requestStatus}
|
||||
dataDir, err := xl.DeleteVersion(del)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wantDir := ""
|
||||
if status == lifecycle.TransitionComplete {
|
||||
wantDir = fi.DataDir
|
||||
}
|
||||
if dataDir != wantDir {
|
||||
t.Errorf("format authorized directory %q, want %q", dataDir, wantDir)
|
||||
}
|
||||
before, err := getFileInfo(raw.Buf, bucket, object, fi.VersionID, fileInfoOpts{Data: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := disk.DeleteVersion(ctx, bucket, object, del, false, DeleteOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
afterRaw, err := disk.ReadXL(ctx, bucket, object, true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
after, err := getFileInfo(afterRaw.Buf, bucket, object, fi.VersionID, fileInfoOpts{Data: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, dirErr := os.ReadFile(filepath.Join(dir, "part.1"))
|
||||
if status != lifecycle.TransitionComplete {
|
||||
if !reflect.DeepEqual(before, after) {
|
||||
t.Error("non-complete cleanup changed metadata or inline data")
|
||||
}
|
||||
if dirErr != nil || !bytes.Equal(data, payload) {
|
||||
t.Errorf("sole local contents removed: %v", dirErr)
|
||||
}
|
||||
} else {
|
||||
if !os.IsNotExist(dirErr) {
|
||||
t.Errorf("completed restore directory still present: %v", dirErr)
|
||||
}
|
||||
if len(after.Data) != 0 || after.Metadata[xhttp.AmzRestore] != "" {
|
||||
t.Error("completed restore kept cached data or restore headers")
|
||||
}
|
||||
if after.TransitionStatus != status || after.VersionID != before.VersionID || after.Metadata["x-amz-meta-preserve"] != "keep" {
|
||||
t.Error("cleanup changed the transitioned version identity or user metadata")
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,453 @@
|
||||
// Copyright (c) 2026 Denis Babchuk
|
||||
//
|
||||
// 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"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||
"github.com/minio/minio/internal/bucket/versioning"
|
||||
xhttp "github.com/minio/minio/internal/http"
|
||||
)
|
||||
|
||||
// restoreTestTier is a remote tier held in memory that counts its reads.
|
||||
type restoreTestTier struct {
|
||||
*tierExpiryTestBackend
|
||||
gets atomic.Int64
|
||||
}
|
||||
|
||||
func (b *restoreTestTier) Get(ctx context.Context, object string, rv remoteVersionID, opts WarmBackendGetOpts) (io.ReadCloser, error) {
|
||||
b.gets.Add(1)
|
||||
return b.tierExpiryTestBackend.Get(ctx, object, rv, opts)
|
||||
}
|
||||
|
||||
// restoreTest is a cluster with a remote tier WARM and a bucket whose
|
||||
// versions are transitioned to it and restored.
|
||||
type restoreTest struct {
|
||||
t *testing.T
|
||||
z *erasureServerPools
|
||||
bucket string
|
||||
object string
|
||||
queue chan expiryOp
|
||||
labels map[string]string // version ID ("null" for the null version) -> label
|
||||
bodies map[string][]byte // label -> data
|
||||
}
|
||||
|
||||
func restoreTestVersionKey(versionID string) string {
|
||||
if versionID == "" {
|
||||
return nullVersionID
|
||||
}
|
||||
return versionID
|
||||
}
|
||||
|
||||
// put writes a version of the object to the first pool. A restored version
|
||||
// is 1 MiB, so that its restored copy has a data directory of its own.
|
||||
func (r *restoreTest) put(label string, null, restored bool) ObjectInfo {
|
||||
r.t.Helper()
|
||||
body := []byte(r.object + " " + label)
|
||||
if restored {
|
||||
body = bytes.Repeat(body, (1<<20)/len(body)+1)
|
||||
}
|
||||
opts := ObjectOptions{Versioned: !null}
|
||||
if null {
|
||||
opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(r.bucket, r.object)
|
||||
}
|
||||
oi, err := r.z.serverPools[0].PutObject(r.t.Context(), r.bucket, r.object,
|
||||
mustGetPutObjReader(r.t, bytes.NewReader(body), int64(len(body)), "", ""), opts)
|
||||
if err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
r.labels[restoreTestVersionKey(oi.VersionID)] = label
|
||||
r.bodies[label] = body
|
||||
return oi
|
||||
}
|
||||
|
||||
// opts returns the options of a request on a version of the object, as the
|
||||
// S3 handlers set them.
|
||||
func (r *restoreTest) opts(versionID string) ObjectOptions {
|
||||
return ObjectOptions{
|
||||
VersionID: versionID,
|
||||
Versioned: globalBucketVersioningSys.PrefixEnabled(r.bucket, r.object),
|
||||
VersionSuspended: globalBucketVersioningSys.PrefixSuspended(r.bucket, r.object),
|
||||
}
|
||||
}
|
||||
|
||||
// restore transitions a version to the tier, restores it, and moves the
|
||||
// expiry date of the restore into the past.
|
||||
func (r *restoreTest) restore(oi ObjectInfo) {
|
||||
r.t.Helper()
|
||||
ctx := r.t.Context()
|
||||
vopts := r.opts(oi.VersionID)
|
||||
topts := vopts
|
||||
topts.MTime = oi.ModTime
|
||||
topts.Transition = TransitionOptions{Tier: "WARM", ETag: oi.ETag}
|
||||
if err := r.z.TransitionObject(ctx, r.bucket, r.object, topts); err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
ropts := vopts
|
||||
ropts.Transition = TransitionOptions{RestoreRequest: &RestoreObjectRequest{Days: 1}}
|
||||
// Through the pool, as with a single pool: with more than one pool the
|
||||
// pools layer holds the object lock that the write of the restored copy
|
||||
// waits for.
|
||||
if err := r.z.serverPools[0].RestoreTransitionedObject(ctx, r.bucket, r.object, ropts); err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
expired := completedRestoreObj(time.Now().Add(-24 * time.Hour).UTC()).String()
|
||||
mopts := vopts
|
||||
mopts.MTime = oi.ModTime
|
||||
mopts.EvalMetadataFn = func(o *ObjectInfo, _ error) (ReplicateDecision, error) {
|
||||
o.UserDefined[xhttp.AmzRestore] = expired
|
||||
return ReplicateDecision{}, nil
|
||||
}
|
||||
if _, err := r.z.PutObjectMetadata(ctx, r.bucket, r.object, mopts); err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// fileInfo returns the version of the object from a drive of the first pool.
|
||||
func (r *restoreTest) fileInfo(versionID string) FileInfo {
|
||||
r.t.Helper()
|
||||
infos, errs := readAllFileInfo(r.t.Context(), r.z.serverPools[0].getHashedSet(r.object).getDisks(), "", r.bucket, r.object, restoreTestVersionKey(versionID), false, false)
|
||||
if errs[0] != nil {
|
||||
r.t.Fatal(errs[0])
|
||||
}
|
||||
return infos[0]
|
||||
}
|
||||
|
||||
// dataDirs returns how many drives of the first pool hold the data
|
||||
// directory of a version.
|
||||
func (r *restoreTest) dataDirs(dataDir string) int {
|
||||
r.t.Helper()
|
||||
n := 0
|
||||
for _, disk := range r.z.serverPools[0].getHashedSet(r.object).getDisks() {
|
||||
_, err := os.Stat(filepath.Join(disk.Endpoint().Path, r.bucket, r.object, dataDir))
|
||||
if err == nil {
|
||||
n++
|
||||
} else if !os.IsNotExist(err) {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// versions returns the versions of the object in each pool, newest first,
|
||||
// and its free versions in all pools.
|
||||
func (r *restoreTest) versions() (views []string, free int) {
|
||||
r.t.Helper()
|
||||
for _, pool := range r.z.serverPools {
|
||||
rf, err := pool.getHashedSet(r.object).getDisks()[0].ReadXL(r.t.Context(), r.bucket, r.object, false)
|
||||
if errors.Is(err, errFileNotFound) {
|
||||
views = append(views, "")
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
fivs, err := getFileInfoVersions(rf.Buf, r.bucket, r.object, false)
|
||||
if err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
var labels []string
|
||||
for _, fi := range fivs.Versions {
|
||||
if fi.Deleted {
|
||||
labels = append(labels, "DM")
|
||||
} else {
|
||||
labels = append(labels, r.labels[restoreTestVersionKey(fi.VersionID)])
|
||||
}
|
||||
}
|
||||
views = append(views, strings.Join(labels, " "))
|
||||
free += len(fivs.FreeVersions)
|
||||
}
|
||||
return views, free
|
||||
}
|
||||
|
||||
// read returns the label and the data of a version, or of the latest version
|
||||
// when versionID is empty, or the error of the read.
|
||||
func (r *restoreTest) read(versionID string) (string, []byte, error) {
|
||||
r.t.Helper()
|
||||
gr, err := r.z.GetObjectNInfo(r.t.Context(), r.bucket, r.object, nil, nil, ObjectOptions{VersionID: versionID})
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
defer gr.Close()
|
||||
data, err := io.ReadAll(gr)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
return r.labels[restoreTestVersionKey(gr.ObjInfo.VersionID)], data, nil
|
||||
}
|
||||
|
||||
// runQueue runs the queued expiries with an expiry worker.
|
||||
func (r *restoreTest) runQueue() {
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
globalExpiryState.Worker(r.queue)
|
||||
}()
|
||||
r.queue <- nil // the worker returns once it ran every expiry queued before
|
||||
<-done
|
||||
}
|
||||
|
||||
// objInfos returns the versions of the object as a drive of the
|
||||
// first pool holds them, as the scanner and the walks of decommission and
|
||||
// the drive heal see them.
|
||||
func (r *restoreTest) objInfos() []ObjectInfo {
|
||||
r.t.Helper()
|
||||
rf, err := r.z.serverPools[0].getHashedSet(r.object).getDisks()[0].ReadXL(r.t.Context(), r.bucket, r.object, false)
|
||||
if err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
fivs, err := getFileInfoVersions(rf.Buf, r.bucket, r.object, false)
|
||||
if err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
versioned := globalBucketVersioningSys.PrefixEnabled(r.bucket, r.object) || globalBucketVersioningSys.PrefixSuspended(r.bucket, r.object)
|
||||
objInfos := make([]ObjectInfo, len(fivs.Versions))
|
||||
for i, fi := range fivs.Versions {
|
||||
objInfos[i] = fi.ToObjectInfo(r.bucket, r.object, versioned)
|
||||
}
|
||||
return objInfos
|
||||
}
|
||||
|
||||
// The restored copy of a transitioned version expires on the expiry date of
|
||||
// its restore. The expiry removes the copy and nothing else: the version
|
||||
// stays where it is in the version stack, transitioned, and is read from the
|
||||
// tier again. Every path that applies ILM runs the expiry: the scanner and a
|
||||
// GET or HEAD queue it for an expiry worker, decommission and the drive heal
|
||||
// run it at once. The expiry of the restored copy of the latest version used
|
||||
// to delete the object instead: a versioned bucket got a delete marker that
|
||||
// hid it. With versioning suspended it went to the null version, removed its
|
||||
// data if there was one and failed if not, and the restored copy stayed.
|
||||
func TestExpireRestoredCopy(t *testing.T) {
|
||||
prevTier := globalTierConfigMgr
|
||||
t.Cleanup(func() { globalTierConfigMgr = prevTier })
|
||||
prevExpiry := globalExpiryState
|
||||
t.Cleanup(func() { globalExpiryState = prevExpiry })
|
||||
|
||||
// Nothing here is old enough to transition; the rule makes the bucket
|
||||
// one with a tier, as restored versions are in.
|
||||
const rules = `<LifecycleConfiguration><Rule><ID>tier</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter>` +
|
||||
`<Transition><Days>1000</Days><StorageClass>WARM</StorageClass></Transition></Rule></LifecycleConfiguration>`
|
||||
|
||||
type version struct {
|
||||
label string
|
||||
null bool // the null version
|
||||
restored bool // transitioned and restored, the restore has expired
|
||||
}
|
||||
testCases := []struct {
|
||||
name string
|
||||
versioning versioning.State // "", Enabled or Suspended
|
||||
versions []version // oldest first
|
||||
}{
|
||||
{name: "versioned", versioning: versioning.Enabled, versions: []version{{label: "v1"}, {label: "v2", restored: true}}},
|
||||
{name: "versioned-noncurrent", versioning: versioning.Enabled, versions: []version{{label: "v1", restored: true}, {label: "v2"}}},
|
||||
{name: "versioned-null", versioning: versioning.Enabled, versions: []version{{label: "v1", null: true, restored: true}}},
|
||||
{name: "suspended", versioning: versioning.Suspended, versions: []version{{label: "v1", null: true}, {label: "v2", restored: true}}},
|
||||
{name: "suspended-no-null", versioning: versioning.Suspended, versions: []version{{label: "v1"}, {label: "v2", restored: true}}},
|
||||
{name: "suspended-null", versioning: versioning.Suspended, versions: []version{{label: "v1"}, {label: "v2", null: true, restored: true}}},
|
||||
{name: "suspended-null-noncurrent", versioning: versioning.Suspended, versions: []version{{label: "v1", null: true, restored: true}, {label: "v2"}}},
|
||||
{name: "unversioned", versions: []version{{label: "v1", null: true, restored: true}}},
|
||||
}
|
||||
|
||||
sources := []struct {
|
||||
name string
|
||||
expire func(r *restoreTest, oi ObjectInfo, evt lifecycle.Event)
|
||||
}{
|
||||
{"scanner", func(r *restoreTest, _ ObjectInfo, _ lifecycle.Event) {
|
||||
lc, err := globalLifecycleSys.Get(r.bucket)
|
||||
if err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
lr, err := globalBucketObjectLockSys.Get(r.bucket)
|
||||
if err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
item := scannerItem{bucket: r.bucket, objectName: r.object, lifeCycle: lc}
|
||||
item.applyActions(r.t.Context(), r.z, r.objInfos(), lr, &sizeSummary{}, nil)
|
||||
r.runQueue()
|
||||
}},
|
||||
{"head", func(r *restoreTest, oi ObjectInfo, evt lifecycle.Event) {
|
||||
applyExpiryRule(evt, lcEventSrc_s3HeadObject, oi)
|
||||
r.runQueue()
|
||||
}},
|
||||
{"list", func(r *restoreTest, _ ObjectInfo, _ lifecycle.Event) {
|
||||
// ListObjects of the key with max-keys=1 reads its latest
|
||||
// version without the versioning of the bucket, a listing of
|
||||
// versions reads them all.
|
||||
if _, err := r.z.ListObjects(r.t.Context(), r.bucket, r.object, "", "", 1); err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
if _, err := r.z.ListObjectVersions(r.t.Context(), r.bucket, r.object, "", "", "", 100); err != nil {
|
||||
r.t.Fatal(err)
|
||||
}
|
||||
r.runQueue()
|
||||
}},
|
||||
{"decom", func(r *restoreTest, oi ObjectInfo, evt lifecycle.Event) {
|
||||
for _, o := range r.objInfos() {
|
||||
if o.VersionID == oi.VersionID {
|
||||
applyExpiryOnTransitionedObject(r.t.Context(), r.z, o, evt, lcEventSrc_Decom)
|
||||
return
|
||||
}
|
||||
}
|
||||
r.t.Fatal("version not found on the drive")
|
||||
}},
|
||||
}
|
||||
|
||||
for _, pools := range []int{1, 2} {
|
||||
t.Run(fmt.Sprintf("pools=%d", pools), func(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
dirs, err := getRandomDisks(4 * pools)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { removeRoots(dirs) })
|
||||
var endpoints EndpointServerPools
|
||||
for i := range pools {
|
||||
endpoints = append(endpoints, mustGetPoolEndpoints(i, dirs[4*i:4*i+4]...)...)
|
||||
}
|
||||
obj, _, err := initObjectLayer(ctx, endpoints)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
z := obj.(*erasureServerPools)
|
||||
t.Cleanup(func() { z.Shutdown(context.Background()) })
|
||||
// After initObjectLayer, which sets up a new tier config.
|
||||
tier := &restoreTestTier{tierExpiryTestBackend: &tierExpiryTestBackend{objects: map[string][]byte{}}}
|
||||
globalTierConfigMgr = NewTierConfigMgr()
|
||||
globalTierConfigMgr.drivercache["WARM"] = tier
|
||||
|
||||
for _, tc := range testCases {
|
||||
for _, src := range sources {
|
||||
t.Run(tc.name+"/"+src.name, func(t *testing.T) {
|
||||
r := &restoreTest{
|
||||
t: t, z: z,
|
||||
bucket: "restore-" + tc.name + "-" + src.name, object: "obj",
|
||||
queue: make(chan expiryOp, 100),
|
||||
labels: map[string]string{}, bodies: map[string][]byte{},
|
||||
}
|
||||
if err := z.MakeBucket(ctx, r.bucket, MakeBucketOptions{VersioningEnabled: tc.versioning != ""}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
meta, err := globalBucketMetadataSys.Get(r.bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tc.versioning == versioning.Suspended {
|
||||
meta.VersioningConfigXML = []byte("<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>")
|
||||
}
|
||||
meta.LifecycleConfigXML = []byte(rules)
|
||||
if err := meta.Save(ctx, z); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
globalBucketMetadataSys.Set(r.bucket, meta)
|
||||
es := &expiryState{ctx: ctx, objAPI: z}
|
||||
workers := []chan expiryOp{r.queue}
|
||||
es.workers.Store(&workers)
|
||||
globalExpiryState = es
|
||||
|
||||
var restored ObjectInfo
|
||||
for _, v := range tc.versions {
|
||||
oi := r.put(v.label, v.null, v.restored)
|
||||
if v.restored {
|
||||
r.restore(oi)
|
||||
restored = oi
|
||||
}
|
||||
}
|
||||
rlabel := r.labels[restoreTestVersionKey(restored.VersionID)]
|
||||
|
||||
// The fixture: the restored copy is on the drives,
|
||||
// its restore has expired, ILM expires it.
|
||||
fi := r.fileInfo(restored.VersionID)
|
||||
if fi.TransitionStatus != lifecycle.TransitionComplete || fi.DataDir == "" || r.dataDirs(fi.DataDir) == 0 {
|
||||
t.Fatalf("fixture: %s transition status %q, restored copy on %d drives", rlabel, fi.TransitionStatus, r.dataDirs(fi.DataDir))
|
||||
}
|
||||
oi, err := z.GetObjectInfo(ctx, r.bucket, r.object, r.opts(restored.VersionID))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
lc, err := globalLifecycleSys.Get(r.bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
lr, err := globalBucketObjectLockSys.Get(r.bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
evt := evalActionFromLifecycle(ctx, *lc, lr, nil, oi)
|
||||
if !evt.Action.DeleteRestored() || oi.RestoreExpires.IsZero() {
|
||||
t.Fatalf("fixture: %s restore expires %v, action %v", rlabel, oi.RestoreExpires, evt.Action)
|
||||
}
|
||||
before, _ := r.versions()
|
||||
for key, label := range r.labels {
|
||||
if _, data, err := r.read(key); err != nil || !bytes.Equal(data, r.bodies[label]) {
|
||||
t.Fatalf("fixture: GET %s: %v", label, err)
|
||||
}
|
||||
}
|
||||
|
||||
src.expire(r, oi, evt)
|
||||
|
||||
after, free := r.versions()
|
||||
if fmt.Sprint(after) != fmt.Sprint(before) || free != 0 {
|
||||
t.Errorf("versions in each pool: before %q, after %q with %d free versions", before, after, free)
|
||||
}
|
||||
if label, _, err := r.read(""); err != nil || label != tc.versions[len(tc.versions)-1].label {
|
||||
t.Errorf("GET reads %q, %v; want %q", label, err, tc.versions[len(tc.versions)-1].label)
|
||||
}
|
||||
if n := r.dataDirs(fi.DataDir); n != 0 {
|
||||
t.Errorf("restored copy of %s still on %d drives", rlabel, n)
|
||||
}
|
||||
oi, err = z.GetObjectInfo(ctx, r.bucket, r.object, r.opts(restored.VersionID))
|
||||
if err != nil {
|
||||
t.Fatalf("HEAD %s: %v", rlabel, err)
|
||||
}
|
||||
if oi.TransitionedObject.Status != lifecycle.TransitionComplete || oi.UserDefined[xhttp.AmzRestore] != "" || !oi.RestoreExpires.IsZero() {
|
||||
t.Errorf("HEAD %s: transition status %q, x-amz-restore %q", rlabel, oi.TransitionedObject.Status, oi.UserDefined[xhttp.AmzRestore])
|
||||
}
|
||||
for _, v := range tc.versions {
|
||||
for key, label := range r.labels {
|
||||
if label != v.label {
|
||||
continue
|
||||
}
|
||||
gets := tier.gets.Load()
|
||||
_, data, err := r.read(key)
|
||||
switch {
|
||||
case err != nil:
|
||||
t.Errorf("GET %s: %v", v.label, err)
|
||||
case !bytes.Equal(data, r.bodies[v.label]):
|
||||
t.Errorf("GET %s reads %d bytes that differ", v.label, len(data))
|
||||
case v.restored && tier.gets.Load() == gets:
|
||||
t.Errorf("GET %s does not read the tier", v.label)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
+14
-3
@@ -611,8 +611,9 @@ func enqueueTransitionImmediate(obj ObjectInfo, src lcEventSrc) {
|
||||
func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *ObjectInfo, lcEvent lifecycle.Event, src lcEventSrc) error {
|
||||
traceFn := globalLifecycleSys.trace(*oi)
|
||||
opts := ObjectOptions{
|
||||
Versioned: globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name),
|
||||
Expiration: ExpirationOptions{Expire: true},
|
||||
Versioned: globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name),
|
||||
VersionSuspended: globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name),
|
||||
Expiration: ExpirationOptions{Expire: true},
|
||||
}
|
||||
if lcEvent.Action.DeleteVersioned() {
|
||||
opts.VersionID = oi.VersionID
|
||||
@@ -622,6 +623,17 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob
|
||||
// delete locally restored copy of object or object version
|
||||
// from the source, while leaving metadata behind. The data on
|
||||
// transitioned tier lies untouched and still accessible
|
||||
//
|
||||
// The restored copy belongs to this version, also when it is the
|
||||
// latest one (DeleteRestoredAction). Without its version ID the
|
||||
// delete applies to the object: a versioned bucket gets a delete
|
||||
// marker that hides it, a suspended one strips the null version.
|
||||
// An empty version ID is the null version, as callers that read
|
||||
// it without the versioning of the bucket leave it.
|
||||
opts.VersionID = oi.VersionID
|
||||
if opts.VersionID == "" {
|
||||
opts.VersionID = nullVersionID
|
||||
}
|
||||
opts.Transition.ExpireRestored = true
|
||||
_, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts)
|
||||
if err == nil {
|
||||
@@ -636,7 +648,6 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob
|
||||
// touching the remote tier. If the lifecycle configuration changed after
|
||||
// the scanner queued this task, DeleteObject rejects the stale action while
|
||||
// the remote object is still protected.
|
||||
opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name)
|
||||
keepRemote := opts.VersionID == "" &&
|
||||
(opts.Versioned || (opts.VersionSuspended && oi.VersionID != "" && oi.VersionID != nullVersionID))
|
||||
if lcEvent.Action == lifecycle.DeleteAction || lcEvent.Action == lifecycle.DeleteVersionAction {
|
||||
|
||||
@@ -2114,7 +2114,9 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
|
||||
tryDel := false
|
||||
if gerr != nil && goi.Name == "" {
|
||||
if _, ok := gerr.(InsufficientReadQuorum); ok {
|
||||
if opts.Versioned || opts.VersionSuspended || countOnlineDisks(storageDisks) < len(storageDisks)/2+1 {
|
||||
// A restored-copy expiry needs fresh metadata to confirm its target
|
||||
// and deadline; ordinary unversioned DELETE may still try a blind delete.
|
||||
if opts.Transition.ExpireRestored || opts.Versioned || opts.VersionSuspended || countOnlineDisks(storageDisks) < len(storageDisks)/2+1 {
|
||||
// Add an MRF heal for next time.
|
||||
er.addPartial(bucket, object, opts.VersionID)
|
||||
return objInfo, InsufficientWriteQuorum{}
|
||||
|
||||
@@ -1498,6 +1498,11 @@ func (x *xlMetaV2) DeleteVersion(fi FileInfo) (string, error) {
|
||||
}
|
||||
switch {
|
||||
case fi.ExpireRestored:
|
||||
// Only a completed transition has a remote copy. Trust the selected
|
||||
// stored version, not the transition status supplied by the caller.
|
||||
if string(ver.ObjectV2.MetaSys[metaTierStatus]) != lifecycle.TransitionComplete {
|
||||
return "", nil
|
||||
}
|
||||
ver.ObjectV2.RemoveRestoreHdrs()
|
||||
err = x.setIdx(i, *ver)
|
||||
case fi.TransitionStatus == lifecycle.TransitionComplete:
|
||||
|
||||
Reference in New Issue
Block a user