diff --git a/cmd/erasure-server-pool-placement_test.go b/cmd/erasure-server-pool-placement_test.go new file mode 100644 index 0000000000000..660e5c0c96452 --- /dev/null +++ b/cmd/erasure-server-pool-placement_test.go @@ -0,0 +1,331 @@ +// 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 . + +package cmd + +import ( + "bytes" + "context" + "errors" + "fmt" + "slices" + "testing" + + "github.com/minio/minio/internal/config/storageclass" +) + +// Report a fixed fill level for allocation only; metadata and data use the +// real fixture disks. 90% is above the reserve that steers new objects to +// other pools but below the hard free drive threshold; 100% is past it. +type placementCapacityDisk struct { + StorageAPI + usedPct uint64 +} + +func (d placementCapacityDisk) DiskInfo(ctx context.Context, opts DiskInfoOptions) (DiskInfo, error) { + info, err := d.StorageAPI.DiskInfo(ctx, opts) + info.Total = 1 << 40 + info.Used = info.Total * d.usedPct / 100 + info.Free = info.Total - info.Used + return info, err +} + +func placementCapacity(z *erasureServerPools, object string, usedPct ...uint64) func() { + var restore []func() + for i, pool := range z.serverPools { + restore = append(restore, conditionalPutSwapDisks(pool, object, func(disk StorageAPI) StorageAPI { + return placementCapacityDisk{StorageAPI: disk, usedPct: usedPct[i]} + })) + } + return func() { + for _, fn := range restore { + fn() + } + } +} + +// Capacity with the object's pool at pct and the other pool empty. +func placementHomeCapacity(z *erasureServerPools, object string, home int, pct uint64) func() { + used := make([]uint64, len(z.serverPools)) + used[home] = pct + return placementCapacity(z, object, used...) +} + +// placementPools returns the pools holding versionID of object, delete +// markers included. An empty versionID addresses the latest version. +func placementPools(t *testing.T, z *erasureServerPools, bucket, object, versionID string) []int { + t.Helper() + var pools []int + for i, pool := range z.serverPools { + oi, err := pool.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: versionID}) + switch { + case err == nil, oi.DeleteMarker: + pools = append(pools, i) + case !isErrObjectNotFound(err) && !isErrVersionNotFound(err): + t.Fatalf("pool %d: %v", i, err) + } + } + return pools +} + +func placementPut(t *testing.T, z *erasureServerPools, bucket, object, body string, opts ObjectOptions) ObjectInfo { + t.Helper() + oi, err := z.PutObject(t.Context(), bucket, object, + mustGetPutObjReader(t, bytes.NewBufferString(body), int64(len(body)), "", ""), opts) + if err != nil { + t.Fatal(err) + } + return oi +} + +func placementMultipartPut(t *testing.T, z *erasureServerPools, bucket, object, body string, opts ObjectOptions) ObjectInfo { + t.Helper() + mp, err := z.NewMultipartUpload(t.Context(), bucket, object, opts) + if err != nil { + t.Fatal(err) + } + part, err := z.PutObjectPart(t.Context(), bucket, object, mp.UploadID, 1, + mustGetPutObjReader(t, bytes.NewBufferString(body), int64(len(body)), "", ""), opts) + if err != nil { + t.Fatal(err) + } + oi, err := z.CompleteMultipartUpload(t.Context(), bucket, object, mp.UploadID, + []CompletePart{{PartNumber: 1, ETag: part.ETag}}, opts) + if err != nil { + t.Fatal(err) + } + return oi +} + +// Take the first n drives of the set that holds object offline. +func placementOffline(pool *erasureSets, object string, n int) func() { + var i int + return conditionalPutSwapDisks(pool, object, func(disk StorageAPI) StorageAPI { + if i++; i <= n { + return nil + } + return disk + }) +} + +func TestPoolsOverwriteStaysInObjectPool(t *testing.T) { + for _, mode := range []string{"unversioned", "versioned", "suspended"} { + t.Run(mode, func(t *testing.T) { + z, bucket := consistencyPools(t) + opts := ObjectOptions{Versioned: mode == "versioned", VersionSuspended: mode == "suspended"} + for home := range z.serverPools { + object := fmt.Sprintf("object-%d", home) + putConsistencyObject(t, z, bucket, object, home, "v0", opts) + // Capacity-weighted selection would always move the object to the + // other pool: only that pool is below the reserve. + defer placementHomeCapacity(z, object, home, 90)() + var versions []string + for i := 1; i <= 5; i++ { + versions = append(versions, placementPut(t, z, bucket, object, fmt.Sprintf("v%d", i), opts).VersionID) + } + if got := placementPools(t, z, bucket, object, ""); !slices.Equal(got, []int{home}) { + t.Errorf("%s was in pool %d, after overwrites it is in pools %v", object, home, got) + continue + } + if mode != "versioned" { + continue + } + for _, versionID := range versions { + if got := placementPools(t, z, bucket, object, versionID); !slices.Equal(got, []int{home}) { + t.Errorf("%s version %s is in pools %v, want [%d]", object, versionID, got, home) + } + } + } + }) + } +} + +// Without room for the write in the object's pool an overwrite still goes to +// another pool, as before, and fails only when no pool has room. +func TestPoolsOverwriteFullObjectPool(t *testing.T) { + z, bucket := consistencyPools(t) + const object = "full" + opts := ObjectOptions{Versioned: true} + putConsistencyObject(t, z, bucket, object, 0, "v0", opts) + + restore := placementHomeCapacity(z, object, 0, 100) + oi := placementPut(t, z, bucket, object, "v1", opts) + restore() + if got := placementPools(t, z, bucket, object, oi.VersionID); !slices.Equal(got, []int{1}) { + t.Fatalf("overwrite with pool 0 full went to pools %v, want [1]", got) + } + + defer placementCapacity(z, object, 100, 100)() + _, err := z.PutObject(t.Context(), bucket, object, + mustGetPutObjReader(t, bytes.NewBufferString("v2"), 2, "", ""), opts) + if !errors.As(err, &StorageFull{}) { + t.Fatalf("overwrite with all pools full: got %v, want StorageFull", err) + } +} + +// An EC 8+8 version stays readable with 8 of 16 drives online, while a write +// needs more of them. The overwrite must go to another pool before the +// request body is read, and the object's set may only report the write quorum +// error when no other pool has room. +func TestPoolsOverwriteObjectSetWithoutWriteQuorum(t *testing.T) { + for _, tc := range []struct { + optimize string + offline int + }{ + // Offline drives raise the parity of the write up to 8+8, which + // needs 9 drives. + {"availability", 8}, + // The write keeps the standard 12+4 and needs 12 drives. + {"capacity", 5}, + } { + t.Run(tc.optimize, func(t *testing.T) { + z, bucket := consistencyPools(t) + previous := globalStorageClass + globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 4}, Optimize: tc.optimize}) + t.Cleanup(func() { globalStorageClass.Update(previous) }) + opts := ObjectOptions{Versioned: true} + for home := range z.serverPools { + other := 1 - home + object, full := fmt.Sprintf("quorum-%d", home), fmt.Sprintf("quorum-full-%d", home) + for _, key := range []string{object, full} { + putConsistencyObject(t, z, bucket, key, home, "v0", ObjectOptions{Versioned: true, MaxParity: true}) + } + + var put, multipart ObjectInfo + func() { + // The other pool is above the reserve, so capacity-weighted + // selection alone would pick the object's pool again. + defer placementHomeCapacity(z, object, other, 90)() + defer placementOffline(z.serverPools[home], object, tc.offline)() + put = placementPut(t, z, bucket, object, "v1", opts) + multipart = placementMultipartPut(t, z, bucket, object, "v2", opts) + }() + for _, versionID := range []string{put.VersionID, multipart.VersionID} { + if got := placementPools(t, z, bucket, object, versionID); !slices.Equal(got, []int{other}) { + t.Errorf("%s version %s is in pools %v, want [%d]", object, versionID, got, other) + } + } + + func() { + defer placementHomeCapacity(z, full, other, 100)() + defer placementOffline(z.serverPools[home], full, tc.offline)() + _, err := z.PutObject(t.Context(), bucket, full, + mustGetPutObjReader(t, bytes.NewBufferString("v1"), 2, "", ""), opts) + if !errors.As(err, &InsufficientWriteQuorum{}) { + t.Errorf("%s overwrite with the other pool full: got %v, want InsufficientWriteQuorum", full, err) + } + _, err = z.NewMultipartUpload(t.Context(), bucket, full, opts) + if !errors.As(err, &InsufficientWriteQuorum{}) { + t.Errorf("%s multipart overwrite with the other pool full: got %v, want InsufficientWriteQuorum", full, err) + } + }() + } + }) + } +} + +// A single pool keeps the free drive threshold from #21115 for overwrites. +func TestSinglePoolOverwriteFreeDriveThreshold(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + obj, dirs, err := prepareErasure16(ctx) + if err != nil { + cancel() + t.Fatal(err) + } + z := obj.(*erasureServerPools) + t.Cleanup(func() { cancel(); z.Shutdown(context.Background()); removeRoots(dirs) }) + bucket := "single-pool-placement" + if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil { + t.Fatal(err) + } + const object = "threshold" + placementPut(t, z, bucket, object, "v0", ObjectOptions{}) + + restore := placementCapacity(z, object, 90) + placementPut(t, z, bucket, object, "v1", ObjectOptions{}) + restore() + + defer placementCapacity(z, object, 100)() + _, err = z.PutObject(t.Context(), bucket, object, + mustGetPutObjReader(t, bytes.NewBufferString("v2"), 2, "", ""), ObjectOptions{}) + if !errors.As(err, &StorageFull{}) { + t.Fatalf("overwrite with the pool full: got %v, want StorageFull", err) + } +} + +// A draining pool does not take new versions, even of objects it holds. +func TestPoolsOverwriteSkipsDrainingPool(t *testing.T) { + for _, rebalance := range []bool{false, true} { + t.Run(fmt.Sprintf("rebalance=%t", rebalance), func(t *testing.T) { + z, bucket := consistencyPools(t) + const object = "draining" + opts := ObjectOptions{Versioned: true} + putConsistencyObject(t, z, bucket, object, 0, "v0", opts) + if rebalance { + z.rebalMu.Lock() + z.rebalMeta = &rebalanceMeta{PoolStats: []*rebalanceStats{ + {Participating: true, Info: rebalanceInfo{Status: rebalStarted}}, {}, + }} + z.rebalMu.Unlock() + } else { + z.poolMetaMutex.Lock() + z.poolMeta.Pools[0].Decommission = &PoolDecommissionInfo{} + z.poolMetaMutex.Unlock() + } + defer placementHomeCapacity(z, object, 1, 90)() + oi := placementPut(t, z, bucket, object, "v1", opts) + if got := placementPools(t, z, bucket, object, oi.VersionID); !slices.Equal(got, []int{1}) { + t.Fatalf("overwrite of an object in a draining pool went to pools %v, want [1]", got) + } + }) + } +} + +// Version history behind a latest delete marker stays in the marker's pool. +func TestPoolsOverwriteAfterDeleteMarker(t *testing.T) { + z, bucket := consistencyPools(t) + opts := ObjectOptions{Versioned: true} + for home := range z.serverPools { + object := fmt.Sprintf("marker-%d", home) + putConsistencyObject(t, z, bucket, object, home, "v0", opts) + defer placementHomeCapacity(z, object, home, 90)() + dm, err := z.DeleteObject(t.Context(), bucket, object, opts) + if err != nil { + t.Fatal(err) + } + oi := placementPut(t, z, bucket, object, "v1", opts) + for _, versionID := range []string{dm.VersionID, oi.VersionID} { + if got := placementPools(t, z, bucket, object, versionID); !slices.Equal(got, []int{home}) { + t.Errorf("%s version %s is in pools %v, want [%d]", object, versionID, got, home) + } + } + } +} + +func TestPoolsMultipartOverwriteStaysInObjectPool(t *testing.T) { + z, bucket := consistencyPools(t) + opts := ObjectOptions{Versioned: true} + for home := range z.serverPools { + object := fmt.Sprintf("multipart-%d", home) + putConsistencyObject(t, z, bucket, object, home, "v0", opts) + defer placementHomeCapacity(z, object, home, 90)() + oi := placementMultipartPut(t, z, bucket, object, "v1", opts) + if got := placementPools(t, z, bucket, object, oi.VersionID); !slices.Equal(got, []int{home}) { + t.Errorf("%s multipart version is in pools %v, want [%d]", object, got, home) + } + } +} diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 79b70d51dff2b..9d8fed45fdd37 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -390,7 +390,16 @@ func (p serverPoolsAvailableSpace) FilterMaxUsed(maxUsed int) { // getAvailablePoolIdx will return an index that can hold size bytes. // -1 is returned if no serverPools have available space for the size given. func (z *erasureServerPools) getAvailablePoolIdx(ctx context.Context, bucket, object string, size int64) int { + return z.getAvailablePoolIdxExcept(ctx, bucket, object, size, -1) +} + +// getAvailablePoolIdxExcept is getAvailablePoolIdx with pool skip left out +// of the choice, and of the reserve filter. skip < 0 leaves out no pool. +func (z *erasureServerPools) getAvailablePoolIdxExcept(ctx context.Context, bucket, object string, size int64, skip int) int { serverPools := z.getServerPoolsAvailableSpace(ctx, bucket, object, size) + if skip >= 0 { + serverPools[skip].Available = 0 + } serverPools.FilterMaxUsed(100 - (100 * diskReserveFraction)) total := serverPools.TotalAvailable() if total == 0 { @@ -640,6 +649,13 @@ func (z *erasureServerPools) getPoolIdx(ctx context.Context, bucket, object stri // getWritePoolIdx keeps the write-allocation policy when the caller already // holds the pools-layer object lock and must not reacquire a set read lock. +// +// Writes to an existing object go to the pool that holds its latest version, +// so versions of one object do not spread over pools and an overwrite does +// not leave a stale copy behind. When that version is readable, the object's +// set must also have write quorum and pass the free drive threshold, or the +// write goes to another pool. The choice is made before the request body is +// read, so the write cannot move once the set rejects it. func (z *erasureServerPools) getWritePoolIdx(ctx context.Context, bucket, object string, size int64, noLock bool) (idx int, err error) { pinfo, _, err := z.getPoolInfoExistingWithOpts(ctx, bucket, object, ObjectOptions{ NoLock: noLock, @@ -652,17 +668,71 @@ func (z *erasureServerPools) getWritePoolIdx(ctx context.Context, bucket, object } idx = pinfo.Index - if isErrObjectNotFound(err) || pinfo.Err == nil { + switch { + case isErrObjectNotFound(err): // will generate a temp object idx = z.getAvailablePoolIdx(ctx, bucket, object, size) - if idx < 0 { - return -1, toObjectErr(errDiskFull) + case pinfo.Err == nil: + writable := z.poolHasWriteQuorum(idx, object) + hasSpace := z.poolHasSpaceFor(ctx, idx, bucket, object, size) + if writable && hasSpace { + return idx, nil + } + home := idx + idx = z.getAvailablePoolIdxExcept(ctx, bucket, object, size, home) + if idx < 0 && hasSpace { + // No other pool has room: the object's set reports the write + // quorum error. + return home, nil } } + if idx < 0 { + return -1, toObjectErr(errDiskFull) + } return idx, nil } +// poolHasWriteQuorum reports whether the set of pool idx that would hold +// object has the online drives a standard class putObject or +// newMultipartUpload needs. A version can stay readable without them: EC 8+8 +// reads with 8 of 16 drives but writes with 9. +func (z *erasureServerPools) poolHasWriteQuorum(idx int, object string) bool { + set := z.serverPools[idx].getHashedSet(object) + disks := set.getDisks() + var online int + for _, disk := range disks { + if disk != nil && disk.IsOnline() { + online++ + } + } + if globalStorageClass.AvailabilityOptimized() { + // Offline drives raise the parity, up to half of the set. + return len(disks)-online < (len(disks)+1)/2 + } + parity := globalStorageClass.GetParityForSC("") + if parity < 0 { + parity = set.defaultParityCount + } + quorum := len(disks) - parity + if quorum == parity { + quorum++ + } + return online >= quorum +} + +// poolHasSpaceFor applies the hard free space checks of +// getServerPoolsAvailableSpace to the set of pool idx that would hold object. +// The reserve that steers new objects to emptier pools does not apply. +func (z *erasureServerPools) poolHasSpaceFor(ctx context.Context, idx int, bucket, object string, size int64) bool { + if isMinioMetaBucketName(bucket) { + return true + } + disks := z.serverPools[idx].getHashedSet(object).getDisks() + ok, err := hasSpaceFor(getDiskInfos(ctx, disks...), size) + return err == nil && ok +} + func (z *erasureServerPools) Shutdown(ctx context.Context) error { g := errgroup.WithNErrs(len(z.serverPools))