diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 26059da18e..5bf8c49d26 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -8,6 +8,7 @@ import ( "io" "math" "net/http" + "path" "sort" "strconv" "strings" @@ -17,6 +18,7 @@ import ( "slices" "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util/wildcard" @@ -60,6 +62,9 @@ func (c *commandFsMergeVolumes) Help() string { -apply needs the shell lock (run "lock" first), so two runs never move the same chunks at once. + + Parts of in-progress S3 multipart uploads are not moved, so a source volume + holding them is not cleared until those uploads complete or are cleaned up. ` } @@ -160,12 +165,29 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer seen := newSourceNeedleCounter() + bucketsPath, err := readFilerBucketsPath(commandEnv) + if err != nil { + return err + } + if err := commandEnv.WithFilerClient(false, func(filerClient filer_pb.SeaweedFilerClient) error { return filer_pb.TraverseBfs(context.Background(), commandEnv, util.FullPath(dir), func(parentPath util.FullPath, entry *filer_pb.Entry) error { if entry.IsDirectory { return nil } entryPath := parentPath.Child(entry.Name) + // A completion may already have read this part's fids and will + // commit the object with them, so moving the chunk would leave the + // object pointing at the deleted source needle. + if isMultipartUploadFile(bucketsPath, parentPath) { + found, rErr := c.recordPlanSources(context.Background(), lookupFn, plan, planCollections, seen, entry.Chunks) + if rErr != nil { + fmt.Printf("skip %s: part of an in-progress multipart upload, manifest unreadable: %v\n", entryPath, rErr) + } else if found { + fmt.Printf("skip %s: part of an in-progress multipart upload\n", entryPath) + } + return nil + } if len(entry.HardLinkId) > 0 { if _, seen := processedHardLinks.LoadOrStore(string(entry.HardLinkId), struct{}{}); seen { // Another sibling already carried the HardLinkId through @@ -283,6 +305,48 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer return nil } +// isMultipartUploadFile reports whether entries in dir are the parts of an +// in-progress multipart upload, //.uploads/. +func isMultipartUploadFile(bucketsPath string, dir util.FullPath) bool { + uploadsDir, _ := dir.DirAndName() + bucketDir, uploads := util.FullPath(uploadsDir).DirAndName() + if uploads != s3_constants.MultipartUploadsFolder { + return false + } + parent, _ := util.FullPath(bucketDir).DirAndName() + return parent == path.Clean(bucketsPath) +} + +// recordPlanSources counts the chunks on plan sources as referenced, including +// those behind a manifest that may reference the plan, and reports whether +// there were any. References read before an unreadable manifest stay counted. +func (c *commandFsMergeVolumes) recordPlanSources(ctx context.Context, lookupFn wdclient.LookupFileIdFunctionType, plan *mergePlan, planCollections map[string]bool, seen *sourceNeedleCounter, chunks []*filer_pb.FileChunk) (bool, error) { + found := false + record := func(chunk *filer_pb.FileChunk) { + if vid := needle.VolumeId(chunk.Fid.GetVolumeId()); plan.isSource(vid) { + seen.record(vid) + found = true + } + } + var manifests []*filer_pb.FileChunk + for _, chunk := range chunks { + record(chunk) + if chunk.IsChunkManifest && c.manifestMayReferencePlan(plan, planCollections, needle.VolumeId(chunk.Fid.GetVolumeId())) { + manifests = append(manifests, chunk) + } + } + if len(manifests) == 0 { + return found, nil + } + dataChunks, manifestChunks, err := filer.ResolveChunkManifest(ctx, lookupFn, manifests, 0, math.MaxInt64, nil) + for _, chunk := range slices.Concat(dataChunks, manifestChunks) { + if !slices.Contains(manifests, chunk) { + record(chunk) + } + } + return found, err +} + // warnUnreferencedSources warns when a plan source's index still holds needles // but no filer entry referenced any during traversal — the merge moves nothing // and the real cleanup is volume.fsck. diff --git a/weed/shell/command_fs_merge_volumes_test.go b/weed/shell/command_fs_merge_volumes_test.go index 37500e2d45..17c71986e0 100644 --- a/weed/shell/command_fs_merge_volumes_test.go +++ b/weed/shell/command_fs_merge_volumes_test.go @@ -13,6 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -513,3 +514,124 @@ func TestUnreferencedTargets(t *testing.T) { }) } } + +func TestIsMultipartUploadFile(t *testing.T) { + cases := []struct { + buckets, dir string + want bool + }{ + {"/buckets", "/buckets/b/.uploads/abc", true}, + {"/", "/b/.uploads/abc", true}, + {"/", "/buckets/b/.uploads/abc", false}, + {"/buckets/", "/buckets/b/.uploads/abc", true}, + {"/data/buckets", "/data/buckets/b/.uploads/abc", true}, + {"/buckets", "/buckets/b/.uploads", false}, + {"/buckets", "/buckets/b/dir/.uploads/abc", false}, + {"/buckets", "/buckets/b/.uploads/abc/sub", false}, + {"/buckets", "/other/b/.uploads/abc", false}, + {"/buckets", "/buckets/b/obj", false}, + } + for _, tc := range cases { + if got := isMultipartUploadFile(tc.buckets, util.FullPath(tc.dir)); got != tc.want { + t.Errorf("isMultipartUploadFile(%q, %q) = %v, want %v", tc.buckets, tc.dir, got, tc.want) + } + } +} + +func TestRecordPlanSources(t *testing.T) { + plan := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{ + needle.VolumeId(203): {needle.VolumeId(187)}, + }} + c := newMergeCmd(250000, + &master_pb.VolumeInformationMessage{Id: 203, Collection: "b"}, + &master_pb.VolumeInformationMessage{Id: 187, Collection: "b"}, + &master_pb.VolumeInformationMessage{Id: 5, Collection: "other"}) + planCollections := map[string]bool{"b": true} + chunk := func(vid uint32, key uint64) *filer_pb.FileChunk { + fid := &filer_pb.FileId{VolumeId: vid, FileKey: key, Cookie: 7} + return &filer_pb.FileChunk{Fid: fid, FileId: needle.NewFileId(needle.VolumeId(vid), key, 7).String(), Size: 10} + } + + blob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{chunk(203, 2)}}) + if err != nil { + t.Fatal(err) + } + volumeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write(blob) + })) + defer volumeServer.Close() + lookupFn := func(_ context.Context, fileId string) ([]string, error) { + return []string{volumeServer.URL + "/" + fileId}, nil + } + manifest := func(vid uint32) *filer_pb.FileChunk { + m := chunk(vid, 1) + m.IsChunkManifest = true + return m + } + + for _, tc := range []struct { + name string + chunks []*filer_pb.FileChunk + found bool + count int + }{ + {"off plan sources", []*filer_pb.FileChunk{chunk(187, 1), chunk(5, 1)}, false, 0}, + {"on a plan source", []*filer_pb.FileChunk{chunk(203, 1), chunk(187, 2), chunk(203, 3)}, true, 2}, + {"behind a manifest off the source", []*filer_pb.FileChunk{manifest(187)}, true, 1}, + {"manifest outside the plan collections", []*filer_pb.FileChunk{manifest(5)}, false, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + seen := newSourceNeedleCounter() + found, err := c.recordPlanSources(context.Background(), lookupFn, plan, planCollections, seen, tc.chunks) + if err != nil { + t.Fatal(err) + } + if found != tc.found || seen.count(needle.VolumeId(203)) != tc.count { + t.Fatalf("found %v with %d references to 203, want %v with %d", found, seen.count(needle.VolumeId(203)), tc.found, tc.count) + } + }) + } + + unreachable := func(_ context.Context, fileId string) ([]string, error) { + return nil, fmt.Errorf("volume of %s unavailable", fileId) + } + seen := newSourceNeedleCounter() + found, err := c.recordPlanSources(context.Background(), unreachable, plan, planCollections, seen, []*filer_pb.FileChunk{chunk(203, 1), manifest(187)}) + if err == nil { + t.Fatal("an unreadable manifest must be reported") + } + if !found || seen.count(needle.VolumeId(203)) != 1 { + t.Fatalf("direct reference lost with an unreadable manifest: found %v, %d references to 203", found, seen.count(needle.VolumeId(203))) + } + + unreadable := chunk(187, 10) + unreadable.IsChunkManifest = true + nestedBlob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{unreadable}}) + if err != nil { + t.Fatal(err) + } + outer := chunk(187, 9) + outer.IsChunkManifest = true + partServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/"+outer.GetFileIdString() { + _, _ = w.Write(nestedBlob) + return + } + _, _ = w.Write(blob) + })) + defer partServer.Close() + failsLater := func(_ context.Context, fileId string) ([]string, error) { + if fileId == unreadable.GetFileIdString() { + return nil, fmt.Errorf("volume of %s unavailable", fileId) + } + return []string{partServer.URL + "/" + fileId}, nil + } + seen = newSourceNeedleCounter() + found, err = c.recordPlanSources(context.Background(), failsLater, plan, planCollections, seen, []*filer_pb.FileChunk{manifest(187), outer}) + if err == nil { + t.Fatal("an unreadable manifest must be reported") + } + if !found || seen.count(needle.VolumeId(203)) != 1 { + t.Fatalf("chunks behind a readable manifest lost when a later one is unreadable: found %v, %d references to 203", found, seen.count(needle.VolumeId(203))) + } +}