mirror of
https://github.com/pgsty/minio.git
synced 2026-11-09 17:21:51 +03:00
fix(decom): move the other buckets when one cannot be finished
A decommission stopped at the first bucket it could not finish. A free version whose tier is no longer configured is enough: its remote object cannot be removed, so the free version stays, the bucket's check for versions left behind fails, and every bucket after it stayed in the pool. decommissionInBackground now keeps the first error, continues with the next bucket, and returns that error after the last one. A canceled context still returns at once. A bucket is retired from the resumable queue only when it is empty, as before, so the decommission still fails and a retry resumes with the buckets that are left. removeFreeVersions now names the free version, its remote object and the tier when the tier is not configured, and says to configure the tier again under the same name. The free version is the only record of its remote object, so it is kept; nothing is dropped on inference. Until the tier is configured again, such a free version blocks the decommission of its pool. Tests: TestPoolsDecommissionContinuesPastStuckBucket and TestRemoveFreeVersionsMissingTierError fail on the base. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
@@ -1098,12 +1098,26 @@ func (m *decomMetrics) log(d decomMetric, poolIdx int, paths ...string) func(z i
|
||||
}
|
||||
}
|
||||
|
||||
// decommissionInBackground decommissions the pending buckets of pool idx. A
|
||||
// bucket that cannot be finished stays queued and does not stop the others:
|
||||
// the first such error is returned after the last bucket, so the
|
||||
// decommission fails and a retry resumes with what is left. A canceled
|
||||
// context returns at once.
|
||||
func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx int) error {
|
||||
pool := z.serverPools[idx]
|
||||
z.poolMetaMutex.RLock()
|
||||
pending := z.poolMeta.PendingBuckets(idx)
|
||||
z.poolMetaMutex.RUnlock()
|
||||
|
||||
var firstErr error
|
||||
failed := func(bucket decomBucketInfo, err error) {
|
||||
if firstErr == nil {
|
||||
firstErr = err
|
||||
return
|
||||
}
|
||||
// Only the first error is returned; log the others.
|
||||
decomLogIf(ctx, fmt.Errorf("decommission of bucket %s: %w", bucket, err))
|
||||
}
|
||||
for _, bucket := range pending {
|
||||
z.poolMetaMutex.RLock()
|
||||
isDecommissioned := z.poolMeta.isBucketDecommissioned(idx, bucket.String())
|
||||
@@ -1127,7 +1141,11 @@ func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx i
|
||||
stopFn := globalDecommissionMetrics.log(decomMetricDecommissionBucket, idx, bucket.Name)
|
||||
if err := z.decommissionPool(ctx, idx, pool, bucket); err != nil {
|
||||
stopFn(0, err)
|
||||
return err
|
||||
if contextCanceled(ctx) {
|
||||
return err
|
||||
}
|
||||
failed(bucket, err)
|
||||
continue
|
||||
}
|
||||
// Do not retire the bucket from the resumable queue until every
|
||||
// source version and free version is gone. In particular, an ILM
|
||||
@@ -1136,7 +1154,11 @@ func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx i
|
||||
// before that action performs its logical and tier cleanup.
|
||||
if err := z.checkAfterDecomBucket(ctx, idx, bucket); err != nil {
|
||||
stopFn(0, err)
|
||||
return err
|
||||
if contextCanceled(ctx) {
|
||||
return err
|
||||
}
|
||||
failed(bucket, err)
|
||||
continue
|
||||
}
|
||||
stopFn(0, nil)
|
||||
|
||||
@@ -1146,7 +1168,7 @@ func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx i
|
||||
}
|
||||
z.poolMetaMutex.Unlock()
|
||||
}
|
||||
return nil
|
||||
return firstErr
|
||||
}
|
||||
|
||||
// removeFreeVersions removes the free versions fvs of object from the pool
|
||||
@@ -1161,6 +1183,13 @@ func (z *erasureServerPools) removeFreeVersions(ctx context.Context, bucket, obj
|
||||
}
|
||||
for _, fv := range fvs {
|
||||
err := deleteObjectFromRemoteTier(ctx, fv.TransitionedObjName, fv.TransitionVersionID, fv.TransitionTier)
|
||||
if errors.Is(err, errTierNotFound) {
|
||||
// The free version is the only record of its remote object. It
|
||||
// stays until the tier can be reached again.
|
||||
return fmt.Errorf("unable to remove the free version %s of %s/%s: its remote object %s is in tier %s, which is not configured; "+
|
||||
"configure tier %s again under the same name so that the remote object and the free version can be removed: %w",
|
||||
fv.VersionID, bucket, object, fv.TransitionedObjName, fv.TransitionTier, fv.TransitionTier, err)
|
||||
}
|
||||
if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
|
||||
return fmt.Errorf("unable to remove the remote object %s of a free version of %s/%s from tier %s: %w",
|
||||
fv.TransitionedObjName, bucket, object, fv.TransitionTier, err)
|
||||
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -524,3 +525,98 @@ func TestFreeVersionDecommissionTierUnavailable(t *testing.T) {
|
||||
t.Errorf("remote objects removed: %v, want %v", f.tier.removed, want)
|
||||
}
|
||||
}
|
||||
|
||||
// A free version whose tier is no longer configured keeps its bucket in the
|
||||
// decommission queue, and the decommission fails, but the buckets after it
|
||||
// are still moved. Once the tier is configured again, a retry completes it.
|
||||
func TestPoolsDecommissionContinuesPastStuckBucket(t *testing.T) {
|
||||
f := newFreeVersionTest(t, "")
|
||||
ctx := t.Context()
|
||||
const other = "zz-other" // listed after f.bucket
|
||||
if err := f.z.MakeBucket(ctx, other, MakeBucketOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
keys := []string{"k1", "k2", "dir/k3"}
|
||||
for _, key := range keys {
|
||||
putConsistencyObject(t, f.z, other, key, 0, key, ObjectOptions{})
|
||||
}
|
||||
delete(globalTierConfigMgr.drivercache, "WARM")
|
||||
|
||||
if info := f.decommission(); info.Complete || !info.Failed {
|
||||
t.Fatalf("decommission with a free version of a removed tier: complete %v, failed %v; want failed", info.Complete, info.Failed)
|
||||
}
|
||||
f.z.poolMetaMutex.RLock()
|
||||
var queued []string
|
||||
for _, b := range f.z.poolMeta.PendingBuckets(0) {
|
||||
queued = append(queued, b.String())
|
||||
}
|
||||
f.z.poolMetaMutex.RUnlock()
|
||||
if !slices.Contains(queued, f.bucket) || slices.Contains(queued, other) {
|
||||
t.Errorf("queued buckets %v, want %s and not %s", queued, f.bucket, other)
|
||||
}
|
||||
for _, key := range keys {
|
||||
if _, err := f.z.serverPools[0].GetObjectInfo(ctx, other, key, ObjectOptions{}); !isErrObjectNotFound(err) {
|
||||
t.Errorf("%s/%s still in pool 0: %v", other, key, err)
|
||||
}
|
||||
if body := migrationTestBody(t, f.z.GetObjectNInfo, other, key, ""); body != key {
|
||||
t.Errorf("%s/%s reads %q", other, key, body)
|
||||
}
|
||||
}
|
||||
for _, object := range []string{"obj-01", "obj-02"} {
|
||||
if got, want := f.stored(0, object), []string{"free"}; !slices.Equal(got, want) {
|
||||
t.Errorf("pool 0 holds %v of %s, want %v", got, object, want)
|
||||
}
|
||||
}
|
||||
if len(f.tier.removed) != 0 {
|
||||
t.Errorf("remote objects removed: %v", f.tier.removed)
|
||||
}
|
||||
|
||||
// See TestFreeVersionDecommissionTierUnavailable.
|
||||
for _, set := range f.z.serverPools[0].sets {
|
||||
for _, disk := range set.getDisks() {
|
||||
if err := disk.MakeVol(ctx, pathJoin(minioMetaBucket, bucketMetaPrefix)); err != nil && !errors.Is(err, errVolumeExists) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
globalTierConfigMgr.drivercache["WARM"] = f.tier
|
||||
if info := f.decommission(); !info.Complete || info.Failed {
|
||||
t.Fatalf("retry: complete %v, failed %v; want complete", info.Complete, info.Failed)
|
||||
}
|
||||
for _, object := range []string{"obj-01", "obj-02"} {
|
||||
if got := f.stored(0, object); len(got) != 0 {
|
||||
t.Errorf("pool 0 holds %v of %s after the retry", got, object)
|
||||
}
|
||||
}
|
||||
if want := []string{f.remote("obj-01", "v3"), f.remote("obj-02", "v1")}; !slices.Equal(slices.Sorted(slices.Values(f.tier.removed)), want) {
|
||||
t.Errorf("remote objects removed: %v, want %v", f.tier.removed, want)
|
||||
}
|
||||
}
|
||||
|
||||
// The error of a free version whose tier is not configured names the free
|
||||
// version, its remote object and the tier, and says what to do.
|
||||
func TestRemoveFreeVersionsMissingTierError(t *testing.T) {
|
||||
f := newFreeVersionTest(t, "")
|
||||
delete(globalTierConfigMgr.drivercache, "WARM")
|
||||
buf := f.xlMeta(0, "obj-02")
|
||||
fivs, err := getFileInfoVersions(buf, f.bucket, "obj-02", false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(fivs.FreeVersions) != 1 {
|
||||
t.Fatalf("%d free versions, want 1", len(fivs.FreeVersions))
|
||||
}
|
||||
fv := fivs.FreeVersions[0]
|
||||
err = f.z.removeFreeVersions(t.Context(), f.bucket, "obj-02", 0, fivs.FreeVersions)
|
||||
if !errors.Is(err, errTierNotFound) {
|
||||
t.Fatalf("err = %v, want errTierNotFound", err)
|
||||
}
|
||||
for _, want := range []string{fv.VersionID, f.remote("obj-02", "v1"), "tier WARM", "configure tier WARM again"} {
|
||||
if !strings.Contains(err.Error(), want) {
|
||||
t.Errorf("error %q does not name %q", err, want)
|
||||
}
|
||||
}
|
||||
if got, want := f.stored(0, "obj-02"), []string{"free"}; !slices.Equal(got, want) {
|
||||
t.Errorf("pool 0 holds %v of obj-02, want %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user