s3: test manifest fold rollback and part range selection
The fold-with-rollback and the boundary-to-byte-range logic were only exercised by hand against a live server; give both an injectable seam and cover the fold, the below-threshold and SSE no-ops, the midway failure deleting the blobs it saved, and offset-vs-legacy-index range selection including indexes that no longer address the chunk list.
This commit is contained in:
Chris Lu
parent
bbb8d48233
commit
47ffa7de96
@ -35,7 +35,10 @@ func (s3a *S3ApiServer) saveManifestChunk(filePath string, bucket string, ttlSec
|
||||
// manifestizeChunks folds a large flat chunk list into manifest chunks. A
|
||||
// failed fold falls back to the flat list, deleting any blobs it saved.
|
||||
func (s3a *S3ApiServer) manifestizeChunks(filePath string, bucket string, ttlSec int32, chunks []*filer_pb.FileChunk) []*filer_pb.FileChunk {
|
||||
save := s3a.saveManifestChunk(filePath, bucket, ttlSec)
|
||||
return manifestizeOrKeepFlat(s3a.saveManifestChunk(filePath, bucket, ttlSec), s3a.deleteOrphanedChunks, filePath, chunks)
|
||||
}
|
||||
|
||||
func manifestizeOrKeepFlat(save filer.SaveDataAsChunkFunctionType, deleteChunks func([]*filer_pb.FileChunk), filePath string, chunks []*filer_pb.FileChunk) []*filer_pb.FileChunk {
|
||||
var saved []*filer_pb.FileChunk
|
||||
record := func(reader io.Reader, name string, offset int64, tsNs int64, expectedDataSize uint64) (*filer_pb.FileChunk, error) {
|
||||
chunk, err := save(reader, name, offset, tsNs, expectedDataSize)
|
||||
@ -48,7 +51,7 @@ func (s3a *S3ApiServer) manifestizeChunks(filePath string, bucket string, ttlSec
|
||||
if err != nil {
|
||||
glog.V(0).Infof("MaybeManifestize %s: %v", filePath, err)
|
||||
if len(saved) > 0 {
|
||||
s3a.deleteOrphanedChunks(saved)
|
||||
deleteChunks(saved)
|
||||
}
|
||||
return chunks
|
||||
}
|
||||
|
||||
134
weed/s3api/s3api_chunk_manifest_test.go
Normal file
134
weed/s3api/s3api_chunk_manifest_test.go
Normal file
@ -0,0 +1,134 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
)
|
||||
|
||||
func flatChunks(n int) []*filer_pb.FileChunk {
|
||||
chunks := make([]*filer_pb.FileChunk, n)
|
||||
for i := range chunks {
|
||||
chunks[i] = &filer_pb.FileChunk{
|
||||
FileId: fmt.Sprintf("1,%x", i+1),
|
||||
Offset: int64(i) * 8,
|
||||
Size: 8,
|
||||
}
|
||||
}
|
||||
return chunks
|
||||
}
|
||||
|
||||
type fakeManifestStore struct {
|
||||
saves int
|
||||
failOn int // 1-based save call to fail on; 0 = never
|
||||
deleted []*filer_pb.FileChunk
|
||||
}
|
||||
|
||||
func (s *fakeManifestStore) save(reader io.Reader, name string, offset int64, tsNs int64, expectedDataSize uint64) (*filer_pb.FileChunk, error) {
|
||||
s.saves++
|
||||
if s.saves == s.failOn {
|
||||
return nil, fmt.Errorf("save %d failed", s.saves)
|
||||
}
|
||||
if _, err := io.Copy(io.Discard, reader); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &filer_pb.FileChunk{FileId: fmt.Sprintf("2,%x", s.saves), Offset: offset}, nil
|
||||
}
|
||||
|
||||
func (s *fakeManifestStore) delete(chunks []*filer_pb.FileChunk) {
|
||||
s.deleted = append(s.deleted, chunks...)
|
||||
}
|
||||
|
||||
func TestManifestizeOrKeepFlatFolds(t *testing.T) {
|
||||
store := &fakeManifestStore{}
|
||||
chunks := flatChunks(filer.ManifestBatch + 50)
|
||||
|
||||
result := manifestizeOrKeepFlat(store.save, store.delete, "/buckets/b/o", chunks)
|
||||
|
||||
manifests, data := filer.SeparateManifestChunks(result)
|
||||
if len(manifests) != 1 || len(data) != 50 {
|
||||
t.Fatalf("expected 1 manifest + 50 flat chunks, got %d + %d", len(manifests), len(data))
|
||||
}
|
||||
if store.saves != 1 || len(store.deleted) != 0 {
|
||||
t.Errorf("expected 1 save and no deletes, got %d saves, %d deleted", store.saves, len(store.deleted))
|
||||
}
|
||||
if manifests[0].Offset != 0 || manifests[0].Size != uint64(filer.ManifestBatch*8) {
|
||||
t.Errorf("manifest span [%d,%d) wrong", manifests[0].Offset, manifests[0].Size)
|
||||
}
|
||||
}
|
||||
|
||||
func TestManifestizeOrKeepFlatBelowThreshold(t *testing.T) {
|
||||
store := &fakeManifestStore{}
|
||||
chunks := flatChunks(filer.ManifestBatch - 1)
|
||||
|
||||
result := manifestizeOrKeepFlat(store.save, store.delete, "/buckets/b/o", chunks)
|
||||
|
||||
if len(result) != len(chunks) || store.saves != 0 {
|
||||
t.Fatalf("expected untouched flat list, got %d chunks, %d saves", len(result), store.saves)
|
||||
}
|
||||
}
|
||||
|
||||
func TestManifestizeOrKeepFlatSkipsSse(t *testing.T) {
|
||||
store := &fakeManifestStore{}
|
||||
chunks := flatChunks(filer.ManifestBatch + 50)
|
||||
chunks[0].SseType = filer_pb.SSEType_SSE_S3
|
||||
|
||||
result := manifestizeOrKeepFlat(store.save, store.delete, "/buckets/b/o", chunks)
|
||||
|
||||
if len(result) != len(chunks) || store.saves != 0 {
|
||||
t.Fatalf("SSE chunks must not be folded, got %d chunks, %d saves", len(result), store.saves)
|
||||
}
|
||||
}
|
||||
|
||||
// A fold that fails midway must fall back to the flat list and delete the
|
||||
// manifest blobs its earlier batches already uploaded.
|
||||
func TestManifestizeOrKeepFlatRollsBackPartialFold(t *testing.T) {
|
||||
store := &fakeManifestStore{failOn: 2}
|
||||
chunks := flatChunks(2*filer.ManifestBatch + 50)
|
||||
|
||||
result := manifestizeOrKeepFlat(store.save, store.delete, "/buckets/b/o", chunks)
|
||||
|
||||
if len(result) != len(chunks) {
|
||||
t.Fatalf("expected fallback to %d flat chunks, got %d", len(chunks), len(result))
|
||||
}
|
||||
if filer.HasChunkManifest(result) {
|
||||
t.Error("fallback list must not contain manifest chunks")
|
||||
}
|
||||
if len(store.deleted) != 1 || store.deleted[0].FileId != "2,1" {
|
||||
t.Fatalf("expected the first saved blob deleted, got %+v", store.deleted)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPartRange(t *testing.T) {
|
||||
chunks := []*filer_pb.FileChunk{
|
||||
{FileId: "1,a", Offset: 0, Size: 8},
|
||||
{FileId: "1,b", Offset: 8, Size: 8},
|
||||
{FileId: "1,c", Offset: 16, Size: 24},
|
||||
}
|
||||
|
||||
// Byte offsets win even when the chunk indexes no longer match the list.
|
||||
start, end, ok := partRange(&PartBoundaryInfo{StartChunk: 40, EndChunk: 80, StartOffset: 16, EndOffset: 40}, chunks)
|
||||
if !ok || start != 16 || end != 39 {
|
||||
t.Errorf("offset boundary: got [%d,%d] ok=%v, want [16,39]", start, end, ok)
|
||||
}
|
||||
|
||||
// Legacy record: range from chunk indexes.
|
||||
start, end, ok = partRange(&PartBoundaryInfo{StartChunk: 1, EndChunk: 3}, chunks)
|
||||
if !ok || start != 8 || end != 39 {
|
||||
t.Errorf("legacy boundary: got [%d,%d] ok=%v, want [8,39]", start, end, ok)
|
||||
}
|
||||
|
||||
// Legacy record with indexes off the list must report, not panic.
|
||||
for _, b := range []*PartBoundaryInfo{
|
||||
{StartChunk: 2, EndChunk: 9},
|
||||
{StartChunk: -1, EndChunk: 2},
|
||||
{StartChunk: 2, EndChunk: 2},
|
||||
} {
|
||||
if _, _, ok := partRange(b, chunks); ok {
|
||||
t.Errorf("boundary %+v should not resolve", b)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -853,19 +853,14 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
|
||||
// Note: ETag is NOT overridden - AWS S3 returns the complete object's ETag
|
||||
// even when requesting a specific part via PartNumber
|
||||
var startOffset, endOffset int64
|
||||
if partInfo != nil && partInfo.EndOffset > partInfo.StartOffset {
|
||||
startOffset = partInfo.StartOffset
|
||||
endOffset = partInfo.EndOffset - 1
|
||||
} else if partInfo != nil {
|
||||
// legacy chunk-index boundaries
|
||||
if partInfo.StartChunk < 0 || partInfo.EndChunk <= partInfo.StartChunk || partInfo.EndChunk > len(objectEntryForSSE.Chunks) {
|
||||
if partInfo != nil {
|
||||
var ok bool
|
||||
startOffset, endOffset, ok = partRange(partInfo, objectEntryForSSE.Chunks)
|
||||
if !ok {
|
||||
glog.Errorf("GetObject: part %d boundary chunks [%d,%d) out of range (chunks: %d)", partNumber, partInfo.StartChunk, partInfo.EndChunk, len(objectEntryForSSE.Chunks))
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
startOffset = objectEntryForSSE.Chunks[partInfo.StartChunk].Offset
|
||||
lastChunk := objectEntryForSSE.Chunks[partInfo.EndChunk-1]
|
||||
endOffset = lastChunk.Offset + int64(lastChunk.Size) - 1
|
||||
} else {
|
||||
// Fallback: assume 1:1 part-to-chunk mapping (backward compatibility)
|
||||
chunkIndex := partNumber - 1
|
||||
@ -3051,6 +3046,20 @@ type PartBoundaryInfo struct {
|
||||
EndOffset int64 `json:"endOffset,omitempty"` // exclusive
|
||||
}
|
||||
|
||||
// partRange returns the inclusive byte range of a part, preferring byte
|
||||
// offsets; ok is false when a legacy record's chunk indexes do not address
|
||||
// the entry's chunk list.
|
||||
func partRange(partInfo *PartBoundaryInfo, chunks []*filer_pb.FileChunk) (startOffset, endOffset int64, ok bool) {
|
||||
if partInfo.EndOffset > partInfo.StartOffset {
|
||||
return partInfo.StartOffset, partInfo.EndOffset - 1, true
|
||||
}
|
||||
if partInfo.StartChunk < 0 || partInfo.EndChunk <= partInfo.StartChunk || partInfo.EndChunk > len(chunks) {
|
||||
return 0, 0, false
|
||||
}
|
||||
lastChunk := chunks[partInfo.EndChunk-1]
|
||||
return chunks[partInfo.StartChunk].Offset, lastChunk.Offset + int64(lastChunk.Size) - 1, true
|
||||
}
|
||||
|
||||
// rc is a helper type that wraps a Reader and Closer for proper resource cleanup
|
||||
type rc struct {
|
||||
io.Reader
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
