mirror of
https://github.com/pgsty/minio.git
synced 2026-11-10 17:45:32 +03:00
Merge heal dry-run fixes (#280)
Takes the dry-run part of PR #280 and credits its author. The parallel deep-verification part stays in #280 for separate review.
This commit is contained in:
@@ -0,0 +1,244 @@
|
|||||||
|
// Copyright (c) 2026 Jasper Bernhardt
|
||||||
|
//
|
||||||
|
// This file is part of Silo Object Storage.
|
||||||
|
//
|
||||||
|
// 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"
|
||||||
|
"crypto/sha256"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"net/url"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"reflect"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/minio/madmin-go/v3"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Unimplemented operations panic: the dry-run tests only allow reads and checks.
|
||||||
|
type nativeAuditDisk struct {
|
||||||
|
StorageAPI
|
||||||
|
info FileInfo
|
||||||
|
readErr error
|
||||||
|
verify func() (*CheckPartsResp, error)
|
||||||
|
deletes atomic.Int32
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *nativeAuditDisk) ReadVersion(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) {
|
||||||
|
return d.info, d.readErr
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *nativeAuditDisk) VerifyFile(context.Context, string, string, FileInfo) (*CheckPartsResp, error) {
|
||||||
|
return d.verify()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *nativeAuditDisk) DeleteVersion(context.Context, string, string, FileInfo, bool, DeleteOptions) error {
|
||||||
|
d.deletes.Add(1)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func nativeAuditMetadata() []FileInfo {
|
||||||
|
meta := make([]FileInfo, 4)
|
||||||
|
for i := range meta {
|
||||||
|
meta[i] = newFileInfo("object", 2, 2)
|
||||||
|
meta[i].Volume, meta[i].Name = "bucket", "object"
|
||||||
|
meta[i].DataDir = "data"
|
||||||
|
meta[i].ModTime = time.Unix(123, 0).UTC()
|
||||||
|
meta[i].Size = 1024
|
||||||
|
meta[i].Erasure.Distribution = []int{1, 2, 3, 4}
|
||||||
|
meta[i].Erasure.Index = i + 1
|
||||||
|
meta[i].Parts = []ObjectPartInfo{{Number: 1, Size: 512}, {Number: 2, Size: 512}}
|
||||||
|
}
|
||||||
|
return meta
|
||||||
|
}
|
||||||
|
|
||||||
|
func nativeAuditEndpoints() []Endpoint {
|
||||||
|
endpoints := make([]Endpoint, 4)
|
||||||
|
for i := range endpoints {
|
||||||
|
endpoints[i].URL = &url.URL{Path: fmt.Sprintf("/synthetic/disk%d", i)}
|
||||||
|
}
|
||||||
|
return endpoints
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNativeAuditDryRunPreservesFaults(t *testing.T) {
|
||||||
|
for _, fault := range []string{"metadata-quorum", "corrupt-metadata", "missing-parts", "corrupt-parts", "inline-corrupt", "delete-marker-quorum"} {
|
||||||
|
t.Run(fault, func(t *testing.T) {
|
||||||
|
meta := nativeAuditMetadata()
|
||||||
|
disks := make([]StorageAPI, len(meta))
|
||||||
|
for i := range disks {
|
||||||
|
disk := &nativeAuditDisk{info: meta[i], verify: func() (*CheckPartsResp, error) {
|
||||||
|
return &CheckPartsResp{Results: []int{checkPartSuccess, checkPartSuccess}}, nil
|
||||||
|
}}
|
||||||
|
if i < 3 {
|
||||||
|
switch fault {
|
||||||
|
case "metadata-quorum", "delete-marker-quorum":
|
||||||
|
disk.readErr = errFileVersionNotFound
|
||||||
|
case "corrupt-metadata":
|
||||||
|
disk.readErr = errFileCorrupt
|
||||||
|
case "missing-parts":
|
||||||
|
disk.verify = func() (*CheckPartsResp, error) { return nil, errFileNotFound }
|
||||||
|
case "corrupt-parts":
|
||||||
|
disk.verify = func() (*CheckPartsResp, error) { return nil, errFileCorrupt }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if fault == "delete-marker-quorum" {
|
||||||
|
disk.info.Deleted = true
|
||||||
|
disk.info.Size = 0
|
||||||
|
}
|
||||||
|
if fault == "inline-corrupt" {
|
||||||
|
disk.info.Data = []byte("corrupt")
|
||||||
|
disk.info.Parts = disk.info.Parts[:1]
|
||||||
|
disk.info.SetInlineData()
|
||||||
|
disk.info.Erasure.Checksums = []ChecksumInfo{{PartNumber: 1, Algorithm: SHA256, Hash: make([]byte, sha256.Size)}}
|
||||||
|
}
|
||||||
|
disks[i] = disk
|
||||||
|
}
|
||||||
|
er := erasureObjects{
|
||||||
|
setDriveCount: 4, defaultParityCount: 2,
|
||||||
|
getDisks: func() []StorageAPI { return disks },
|
||||||
|
getEndpoints: nativeAuditEndpoints,
|
||||||
|
}
|
||||||
|
result, err := er.HealObject(t.Context(), "bucket", "object", "version", madmin.HealOpts{DryRun: true, NoLock: true, Remove: true, ScanMode: madmin.HealDeepScan})
|
||||||
|
for _, disk := range disks {
|
||||||
|
if got := disk.(*nativeAuditDisk).deletes.Load(); got != 0 {
|
||||||
|
t.Fatalf("dry-run deleted data %d times", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
quorumFault := fault == "metadata-quorum" || fault == "corrupt-metadata" || fault == "delete-marker-quorum"
|
||||||
|
if quorumFault && err == nil {
|
||||||
|
t.Fatal("metadata quorum failure reported success")
|
||||||
|
}
|
||||||
|
if !quorumFault && err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(result.Before.Drives) != len(disks) || !reflect.DeepEqual(result.Before, result.After) {
|
||||||
|
t.Fatalf("dry-run states must preserve observed damage: %+v", result)
|
||||||
|
}
|
||||||
|
switch fault {
|
||||||
|
case "metadata-quorum", "delete-marker-quorum":
|
||||||
|
if !isErrVersionNotFound(err) {
|
||||||
|
t.Fatalf("dry-run must return the reduced metadata error, got %v", err)
|
||||||
|
}
|
||||||
|
case "corrupt-metadata":
|
||||||
|
if !errors.Is(err, errFileCorrupt) {
|
||||||
|
t.Fatalf("dry-run must return the reduced metadata error, got %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
unhealthy := 0
|
||||||
|
for _, drive := range result.Before.Drives {
|
||||||
|
if drive.State != madmin.DriveStateOk {
|
||||||
|
unhealthy++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if unhealthy < 3 {
|
||||||
|
t.Fatalf("faults were not reported: %+v", result.Before)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// This control proves the missing-metadata fixture reaches dangling deletion.
|
||||||
|
func TestNativeAuditNonDryRunStillDeletesDangling(t *testing.T) {
|
||||||
|
meta := nativeAuditMetadata()
|
||||||
|
disks := make([]StorageAPI, len(meta))
|
||||||
|
for i := range disks {
|
||||||
|
disk := &nativeAuditDisk{info: meta[i]}
|
||||||
|
if i < 3 {
|
||||||
|
disk.readErr = errFileVersionNotFound
|
||||||
|
}
|
||||||
|
disks[i] = disk
|
||||||
|
}
|
||||||
|
er := erasureObjects{
|
||||||
|
setDriveCount: 4, defaultParityCount: 2,
|
||||||
|
getDisks: func() []StorageAPI { return disks },
|
||||||
|
getEndpoints: nativeAuditEndpoints,
|
||||||
|
}
|
||||||
|
_, err := er.HealObject(t.Context(), "bucket", "object", "version", madmin.HealOpts{NoLock: true, Remove: true, ScanMode: madmin.HealDeepScan})
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("dangling deletion must report the object removed")
|
||||||
|
}
|
||||||
|
for _, disk := range disks {
|
||||||
|
if got := disk.(*nativeAuditDisk).deletes.Load(); got != 1 {
|
||||||
|
t.Fatalf("control did not reach dangling deletion: %d", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNativeAuditDryRunPreservesDanglingFiles(t *testing.T) {
|
||||||
|
obj, roots, err := prepareErasure(t.Context(), 4)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
setObjectLayer(obj)
|
||||||
|
defer removeRoots(roots)
|
||||||
|
defer obj.Shutdown(t.Context())
|
||||||
|
const bucket, object = "native-audit", "dangling-object"
|
||||||
|
if err = obj.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
data := bytes.Repeat([]byte("x"), 1024*1024)
|
||||||
|
if _, err = obj.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
for _, root := range roots[:3] {
|
||||||
|
if err = os.Remove(filepath.Join(root, bucket, object, xlStorageFormatFile)); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
snapshot := func() map[string]string {
|
||||||
|
files := make(map[string]string)
|
||||||
|
for _, root := range roots {
|
||||||
|
err := filepath.Walk(filepath.Join(root, bucket, object), func(path string, info os.FileInfo, err error) error {
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if info.IsDir() {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
data, err := os.ReadFile(path)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
files[path] = fmt.Sprintf("%d/%d/%x", info.Size(), info.ModTime().UnixNano(), sha256.Sum256(data))
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return files
|
||||||
|
}
|
||||||
|
before := snapshot()
|
||||||
|
if len(before) < 2 {
|
||||||
|
t.Fatal("fixture has no surviving metadata and payload files")
|
||||||
|
}
|
||||||
|
result, err := obj.HealObject(t.Context(), bucket, object, "", madmin.HealOpts{DryRun: true, Remove: true, ScanMode: madmin.HealDeepScan})
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("missing metadata quorum reported success")
|
||||||
|
}
|
||||||
|
if !reflect.DeepEqual(result.Before, result.After) {
|
||||||
|
t.Fatalf("dry-run reported changes: %+v", result)
|
||||||
|
}
|
||||||
|
if after := snapshot(); !reflect.DeepEqual(before, after) {
|
||||||
|
t.Fatalf("dry-run changed surviving object files: before=%v after=%v", before, after)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -343,6 +343,11 @@ func (er *erasureObjects) healObject(ctx context.Context, bucket string, object
|
|||||||
|
|
||||||
readQuorum, _, err := objectQuorumFromMeta(ctx, partsMetadata, errs, er.defaultParityCount)
|
readQuorum, _, err := objectQuorumFromMeta(ctx, partsMetadata, errs, er.defaultParityCount)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
if dryRun {
|
||||||
|
// A metadata quorum failure must not delete dangling data in dry-run.
|
||||||
|
return er.defaultHealResult(FileInfo{}, storageDisks, storageEndpoints,
|
||||||
|
errs, bucket, object, versionID), err
|
||||||
|
}
|
||||||
m, derr := er.deleteIfDangling(ctx, bucket, object, partsMetadata, errs, nil, ObjectOptions{
|
m, derr := er.deleteIfDangling(ctx, bucket, object, partsMetadata, errs, nil, ObjectOptions{
|
||||||
VersionID: versionID,
|
VersionID: versionID,
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -1292,8 +1292,10 @@ func (s *peerRESTServer) HealBucketHandler(mss *grid.MSS) (np *grid.JSON[madmin.
|
|||||||
}
|
}
|
||||||
|
|
||||||
bucketDeleted := mss.Get(peerS3BucketDeleted) == "true"
|
bucketDeleted := mss.Get(peerS3BucketDeleted) == "true"
|
||||||
|
dryRun := mss.Get(peerS3BucketDryRun) == "true"
|
||||||
res, err := healBucketLocal(context.Background(), bucket, madmin.HealOpts{
|
res, err := healBucketLocal(context.Background(), bucket, madmin.HealOpts{
|
||||||
Remove: bucketDeleted,
|
Remove: bucketDeleted,
|
||||||
|
DryRun: dryRun,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return np, grid.NewRemoteErr(err)
|
return np, grid.NewRemoteErr(err)
|
||||||
|
|||||||
+11
-4
@@ -354,10 +354,7 @@ func (client *remotePeerS3Client) HealBucket(ctx context.Context, bucket string,
|
|||||||
return madmin.HealResultItem{}, nil
|
return madmin.HealResultItem{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
mss := grid.NewMSSWith(map[string]string{
|
mss := healBucketPeerMSS(bucket, opts)
|
||||||
peerS3Bucket: bucket,
|
|
||||||
peerS3BucketDeleted: strconv.FormatBool(opts.Remove),
|
|
||||||
})
|
|
||||||
|
|
||||||
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
|
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
@@ -366,6 +363,16 @@ func (client *remotePeerS3Client) HealBucket(ctx context.Context, bucket string,
|
|||||||
return resp.ValueOrZero(), toStorageErr(err)
|
return resp.ValueOrZero(), toStorageErr(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// healBucketPeerMSS encodes the bucket heal options a peer applies to its
|
||||||
|
// local drives. Peers that predate the dry-run key ignore it.
|
||||||
|
func healBucketPeerMSS(bucket string, opts madmin.HealOpts) *grid.MSS {
|
||||||
|
return grid.NewMSSWith(map[string]string{
|
||||||
|
peerS3Bucket: bucket,
|
||||||
|
peerS3BucketDeleted: strconv.FormatBool(opts.Remove),
|
||||||
|
peerS3BucketDryRun: strconv.FormatBool(opts.DryRun),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
// GetBucketInfo returns bucket stat info from a peer
|
// GetBucketInfo returns bucket stat info from a peer
|
||||||
func (client *remotePeerS3Client) GetBucketInfo(ctx context.Context, bucket string, opts BucketOptions) (BucketInfo, error) {
|
func (client *remotePeerS3Client) GetBucketInfo(ctx context.Context, bucket string, opts BucketOptions) (BucketInfo, error) {
|
||||||
conn := client.gridConn()
|
conn := client.gridConn()
|
||||||
|
|||||||
@@ -0,0 +1,107 @@
|
|||||||
|
// 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 (
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/minio/madmin-go/v3"
|
||||||
|
"github.com/minio/minio/internal/grid"
|
||||||
|
)
|
||||||
|
|
||||||
|
// A remote node must apply the caller's dry run to its local bucket heal.
|
||||||
|
// The request goes from remotePeerS3Client.HealBucket over a grid connection
|
||||||
|
// to the peer handler, as it does between the nodes of a deployment.
|
||||||
|
func TestPeerHealBucketDryRun(t *testing.T) {
|
||||||
|
obj, roots, err := prepareErasure(t.Context(), 4)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
setObjectLayer(obj)
|
||||||
|
defer removeRoots(roots)
|
||||||
|
defer obj.Shutdown(t.Context())
|
||||||
|
|
||||||
|
tg, err := grid.SetupTestGrid(2)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer tg.Cleanup()
|
||||||
|
if err = healBucketRPC.Register(tg.Managers[1], (&peerRESTServer{}).HealBucketHandler); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
conn := func() *grid.Connection { return tg.Managers[0].Connection(tg.Hosts[1]) }
|
||||||
|
client := &remotePeerS3Client{gridConn: conn}
|
||||||
|
|
||||||
|
const bucket = "peer-heal-dry-run"
|
||||||
|
if err = obj.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
dirs := make([]string, len(roots))
|
||||||
|
for i, root := range roots {
|
||||||
|
dirs[i] = filepath.Join(root, bucket)
|
||||||
|
}
|
||||||
|
if err = os.Remove(dirs[0]); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
exists := func(dir string) bool {
|
||||||
|
_, err := os.Stat(dir)
|
||||||
|
return err == nil
|
||||||
|
}
|
||||||
|
heal := func(opts madmin.HealOpts) {
|
||||||
|
t.Helper()
|
||||||
|
if _, err := client.HealBucket(t.Context(), bucket, opts); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Dry run must not recreate the missing bucket directory.
|
||||||
|
heal(madmin.HealOpts{DryRun: true})
|
||||||
|
if exists(dirs[0]) {
|
||||||
|
t.Fatal("dry-run peer bucket heal created a missing bucket directory")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Dry run must not remove the empty bucket directories.
|
||||||
|
heal(madmin.HealOpts{DryRun: true, Remove: true})
|
||||||
|
for _, dir := range dirs[1:] {
|
||||||
|
if !exists(dir) {
|
||||||
|
t.Fatalf("dry-run peer bucket heal removed %s", dir)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Controls. A message without the dry-run key, as sent by an older node,
|
||||||
|
// still heals: the fixture reaches MakeVol.
|
||||||
|
if _, err = healBucketRPC.Call(t.Context(), conn(), grid.NewMSSWith(map[string]string{
|
||||||
|
peerS3Bucket: bucket,
|
||||||
|
peerS3BucketDeleted: "false",
|
||||||
|
})); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if !exists(dirs[0]) {
|
||||||
|
t.Fatal("control: peer bucket heal did not recreate the missing directory")
|
||||||
|
}
|
||||||
|
|
||||||
|
// A real heal with Remove reaches DeleteVol for the empty directories.
|
||||||
|
heal(madmin.HealOpts{Remove: true})
|
||||||
|
for _, dir := range dirs {
|
||||||
|
if exists(dir) {
|
||||||
|
t.Fatalf("control: peer bucket heal kept empty directory %s", dir)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -29,6 +29,7 @@ import (
|
|||||||
const (
|
const (
|
||||||
peerS3Bucket = "bucket"
|
peerS3Bucket = "bucket"
|
||||||
peerS3BucketDeleted = "bucket-deleted"
|
peerS3BucketDeleted = "bucket-deleted"
|
||||||
|
peerS3BucketDryRun = "dry-run"
|
||||||
peerS3BucketForceCreate = "force-create"
|
peerS3BucketForceCreate = "force-create"
|
||||||
peerS3BucketForceDelete = "force-delete"
|
peerS3BucketForceDelete = "force-delete"
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user