mirror of
https://github.com/pgsty/minio.git
synced 2026-10-28 19:47:23 +03:00
test: cover versioned and encrypted multi-pool recovery
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
@@ -20,13 +20,18 @@ package cmd
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"crypto/sha256"
|
||||
"encoding/base64"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
xhttp "github.com/minio/minio/internal/http"
|
||||
"github.com/minio/minio/internal/kms"
|
||||
)
|
||||
|
||||
// An acknowledged recovery overwrite must resolve the unreadable pool, even
|
||||
@@ -34,74 +39,103 @@ import (
|
||||
// Real inline metadata is combined in the same 15-old/1-new state used by
|
||||
// that regression. Capacity adapters only make destination choice deterministic.
|
||||
func TestPoolsMinorityOverwriteRecovery(t *testing.T) {
|
||||
for _, mode := range []string{"single-pool", "dual-pool-only-inconsistent-copy", "dual-pool-readable-other-copy"} {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
var z *erasureServerPools
|
||||
if mode == "single-pool" {
|
||||
z, _ = mutationPools(t)
|
||||
} else {
|
||||
z, _ = consistencyPools(t)
|
||||
}
|
||||
const object = "review-retry"
|
||||
for _, pool := range z.serverPools {
|
||||
restore := conditionalPutSwapDisks(pool, object, func(d StorageAPI) StorageAPI { return conditionalPutCapacityDisk{StorageAPI: d} })
|
||||
defer restore()
|
||||
}
|
||||
bucket, router := conditionalPutBucket(t, z, "unversioned")
|
||||
oldTime := time.Now().UTC().Add(-time.Hour)
|
||||
oldBody := bytes.Repeat([]byte{'a'}, 8192)
|
||||
pendingBody := bytes.Repeat([]byte{'b'}, 8192)
|
||||
recoveryBody := bytes.Repeat([]byte{'c'}, 8192)
|
||||
seed := func(pool int, body []byte, when time.Time) {
|
||||
t.Helper()
|
||||
putConsistencyObject(t, z, bucket, object, pool, string(body), ObjectOptions{MTime: when})
|
||||
}
|
||||
seed(0, oldBody, oldTime)
|
||||
disks := z.serverPools[0].getHashedSet(object).getDisks()
|
||||
oldMeta := make([][]byte, len(disks))
|
||||
for i, disk := range disks {
|
||||
oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
|
||||
}
|
||||
seed(0, pendingBody, oldTime.Add(time.Minute))
|
||||
for i := 0; i < len(disks)-1; i++ {
|
||||
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if mode == "dual-pool-readable-other-copy" {
|
||||
seed(1, oldBody, oldTime.Add(-time.Minute))
|
||||
// Ordinary allocation can select this healthy other pool.
|
||||
// Force that permitted choice without altering metadata reads.
|
||||
restore := conditionalPutPool(t, z, object, 1)
|
||||
defer restore()
|
||||
}
|
||||
before := consistencyRequest(t, router, http.MethodGet, bucket, object, "")
|
||||
if before.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("fixture must refuse ambiguous read; got %d", before.Code)
|
||||
}
|
||||
req, err := newTestSignedRequestV4(http.MethodPut, getGetObjectURL("", bucket, object), int64(len(recoveryBody)), bytes.NewReader(recoveryBody), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result := httptest.NewRecorder()
|
||||
router.ServeHTTP(result, req.WithContext(t.Context()))
|
||||
t.Logf("recovery PUT HTTP=%d response=%s", result.Code, result.Body.String())
|
||||
if result.Code != http.StatusOK {
|
||||
t.Fatalf("unconditional recovery PUT: %d %s", result.Code, result.Body.String())
|
||||
}
|
||||
after := consistencyRequest(t, router, http.MethodGet, bucket, object, "")
|
||||
t.Logf("after acknowledged recovery PUT: GET HTTP=%d matches_new=%t", after.Code, bytes.Equal(after.Body.Bytes(), recoveryBody))
|
||||
for i, pool := range z.serverPools {
|
||||
oi, err := pool.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{})
|
||||
t.Logf("pool=%d mtime=%s err=%v", i, oi.ModTime, err)
|
||||
}
|
||||
if after.Code != http.StatusOK || !bytes.Equal(after.Body.Bytes(), recoveryBody) {
|
||||
t.Fatalf("acknowledged recovery overwrite did not restore reads: HTTP=%d body=%s", after.Code, after.Body.String())
|
||||
}
|
||||
for _, versioning := range []string{"unversioned", "versioned", "suspended"} {
|
||||
for _, mode := range []string{"single-pool", "dual-pool-only-inconsistent-copy", "dual-pool-readable-other-copy"} {
|
||||
t.Run(versioning+"/"+mode, func(t *testing.T) {
|
||||
testPoolsMinorityOverwriteRecovery(t, mode, versioning, "")
|
||||
})
|
||||
}
|
||||
}
|
||||
for _, encryption := range []string{"SSE-C", "SSE-S3", "SSE-KMS"} {
|
||||
t.Run(encryption, func(t *testing.T) {
|
||||
testPoolsMinorityOverwriteRecovery(t, "dual-pool-readable-other-copy", "unversioned", encryption)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func testPoolsMinorityOverwriteRecovery(t *testing.T, mode, versioning, encryption string) {
|
||||
t.Helper()
|
||||
var z *erasureServerPools
|
||||
if mode == "single-pool" {
|
||||
z, _ = mutationPools(t)
|
||||
} else {
|
||||
z, _ = consistencyPools(t)
|
||||
}
|
||||
const object = "review-retry"
|
||||
for _, pool := range z.serverPools {
|
||||
restore := conditionalPutSwapDisks(pool, object, func(d StorageAPI) StorageAPI { return conditionalPutCapacityDisk{StorageAPI: d} })
|
||||
defer restore()
|
||||
}
|
||||
bucket, router := conditionalPutBucket(t, z, versioning)
|
||||
oldTime := time.Now().UTC().Add(-time.Hour)
|
||||
oldBody := bytes.Repeat([]byte{'a'}, 8192)
|
||||
pendingBody := bytes.Repeat([]byte{'b'}, 8192)
|
||||
recoveryBody := bytes.Repeat([]byte{'c'}, 8192)
|
||||
seed := func(pool int, body []byte, when time.Time) {
|
||||
t.Helper()
|
||||
putConsistencyObject(t, z, bucket, object, pool, string(body), ObjectOptions{MTime: when})
|
||||
}
|
||||
seed(0, oldBody, oldTime)
|
||||
disks := z.serverPools[0].getHashedSet(object).getDisks()
|
||||
oldMeta := make([][]byte, len(disks))
|
||||
for i, disk := range disks {
|
||||
oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
|
||||
}
|
||||
seed(0, pendingBody, oldTime.Add(time.Minute))
|
||||
for i := 0; i < len(disks)-1; i++ {
|
||||
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if mode == "dual-pool-readable-other-copy" {
|
||||
seed(1, oldBody, oldTime.Add(-time.Minute))
|
||||
// Ordinary allocation can select this healthy other pool.
|
||||
// Force that permitted choice without altering metadata reads.
|
||||
restore := conditionalPutPool(t, z, object, 1)
|
||||
defer restore()
|
||||
}
|
||||
before := consistencyRequest(t, router, http.MethodGet, bucket, object, "")
|
||||
if before.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("fixture must refuse ambiguous read; got %d", before.Code)
|
||||
}
|
||||
var headers, readHeaders map[string]string
|
||||
if encryption != "" {
|
||||
oldKMS, oldTLS := GlobalKMS, globalIsTLS
|
||||
GlobalKMS, globalIsTLS = kms.NewStub("recovery-key"), true
|
||||
defer func() { GlobalKMS, globalIsTLS = oldKMS, oldTLS }()
|
||||
headers = map[string]string{xhttp.AmzServerSideEncryption: xhttp.AmzEncryptionAES}
|
||||
switch encryption {
|
||||
case "SSE-C":
|
||||
key := bytes.Repeat([]byte{0x42}, 32)
|
||||
digest := md5.Sum(key)
|
||||
headers = map[string]string{
|
||||
xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
|
||||
xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key),
|
||||
xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(digest[:]),
|
||||
}
|
||||
readHeaders = headers
|
||||
case "SSE-KMS":
|
||||
headers[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionKMS
|
||||
headers[xhttp.AmzServerSideEncryptionKmsID] = "recovery-key"
|
||||
}
|
||||
}
|
||||
url := getPutObjectURL("", bucket, object)
|
||||
result := multipartConditionRequest(t, router, http.MethodPut, url, string(recoveryBody), headers)
|
||||
t.Logf("recovery PUT HTTP=%d response=%s", result.Code, result.Body.String())
|
||||
if result.Code != http.StatusOK {
|
||||
t.Fatalf("unconditional recovery PUT: %d %s", result.Code, result.Body.String())
|
||||
}
|
||||
after := multipartConditionRequest(t, router, http.MethodGet, url, "", readHeaders)
|
||||
t.Logf("after acknowledged recovery PUT: GET HTTP=%d matches_new=%t", after.Code, bytes.Equal(after.Body.Bytes(), recoveryBody))
|
||||
for i, pool := range z.serverPools {
|
||||
oi, err := pool.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{})
|
||||
t.Logf("pool=%d mtime=%s err=%v", i, oi.ModTime, err)
|
||||
}
|
||||
if after.Code != http.StatusOK || !bytes.Equal(after.Body.Bytes(), recoveryBody) {
|
||||
t.Fatalf("acknowledged recovery overwrite did not restore reads: HTTP=%d body=%s", after.Code, after.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// Read failures activated by a completed disk mutation verify that a recovery
|
||||
// ACK is not issued from preflight information alone.
|
||||
type recoveryReadFaultDisk struct {
|
||||
|
||||
Reference in New Issue
Block a user