diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 488006771..b992f811a 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -146,7 +146,9 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer } } - return commandEnv.WithFilerClient(false, func(filerClient filer_pb.SeaweedFilerClient) error { + seen := newSourceNeedleCounter() + + 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 @@ -176,7 +178,10 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer } oldManifestFid := chunk.GetFileIdString() oldManifestVid := chunk.Fid.VolumeId - newChunk, changed, rewritten, mErr := c.rewriteManifestChunk(context.Background(), commandEnv, lookupFn, plan, entryPath, chunk, *apply) + if vid := needle.VolumeId(oldManifestVid); plan.isSource(vid) { + seen.record(vid) + } + newChunk, changed, rewritten, mErr := c.rewriteManifestChunk(context.Background(), commandEnv, lookupFn, plan, entryPath, chunk, *apply, seen.record) if mErr != nil { fmt.Printf("failed to rewrite manifest %s(%s): %v\n", entryPath, oldManifestFid, mErr) continue @@ -199,6 +204,7 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer if !plan.isSource(chunkVolumeId) { continue } + seen.record(chunkVolumeId) oldFid := chunk.GetFileIdString() oldVid := chunk.Fid.VolumeId @@ -234,7 +240,51 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer } return nil }) - }) + }); err != nil { + return err + } + + c.warnUnreferencedSources(writer, plan, seen, dir) + return nil +} + +// 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. +func (c *commandFsMergeVolumes) warnUnreferencedSources(writer io.Writer, plan *mergePlan, seen *sourceNeedleCounter, dir string) { + for src := range plan.targets { + if seen.count(src) > 0 { + continue + } + info := c.volumes[src] + if info == nil || info.FileCount == 0 { + continue + } + fmt.Fprintf(writer, "warning: volume %d has %d needle(s) in its index but no filer entries reference them under %s — nothing merged (orphan needles? run volume.fsck)\n", src, info.FileCount, dir) + } +} + +// sourceNeedleCounter records plan-source needles seen during filer traversal. +// TraverseBfs runs callbacks on concurrent workers, so record is mutex-guarded. +type sourceNeedleCounter struct { + mu sync.Mutex + seen map[needle.VolumeId]int +} + +func newSourceNeedleCounter() *sourceNeedleCounter { + return &sourceNeedleCounter{seen: make(map[needle.VolumeId]int)} +} + +func (s *sourceNeedleCounter) record(vid needle.VolumeId) { + s.mu.Lock() + s.seen[vid]++ + s.mu.Unlock() +} + +func (s *sourceNeedleCounter) count(vid needle.VolumeId) int { + s.mu.Lock() + defer s.mu.Unlock() + return s.seen[vid] } // orphanedNeedle is a needle no filer entry references any more: either a @@ -671,6 +721,7 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk( entryPath util.FullPath, chunk *filer_pb.FileChunk, apply bool, + recordSeen func(needle.VolumeId), ) (*filer_pb.FileChunk, bool, rewrittenNeedles, error) { if !chunk.IsChunkManifest { return chunk, false, rewrittenNeedles{}, fmt.Errorf("not a manifest chunk: %s", chunk.GetFileIdString()) @@ -698,7 +749,10 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk( if sub.IsChunkManifest { oldSubManifestFid := sub.GetFileIdString() oldSubManifestVid := sub.Fid.VolumeId - newSub, changed, nested, rErr := c.rewriteManifestChunk(ctx, commandEnv, lookupFn, plan, entryPath, sub, apply) + if vid := needle.VolumeId(oldSubManifestVid); plan.isSource(vid) { + recordSeen(vid) + } + newSub, changed, nested, rErr := c.rewriteManifestChunk(ctx, commandEnv, lookupFn, plan, entryPath, sub, apply, recordSeen) if rErr != nil { abandon() return chunk, false, rewrittenNeedles{}, rErr @@ -722,6 +776,7 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk( if !plan.isSource(subVid) { continue } + recordSeen(subVid) oldSubFid := sub.GetFileIdString() oldSubVid := sub.Fid.VolumeId toVid, ok := plan.allocate(subVid, sub.Size) diff --git a/weed/shell/command_fs_merge_volumes_test.go b/weed/shell/command_fs_merge_volumes_test.go index 3582692c7..2fe85923a 100644 --- a/weed/shell/command_fs_merge_volumes_test.go +++ b/weed/shell/command_fs_merge_volumes_test.go @@ -2,6 +2,7 @@ package shell import ( "strings" + "sync" "testing" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" @@ -271,3 +272,77 @@ func TestManifestMayReferencePlan(t *testing.T) { } } } + +// A source whose index still holds needles but that the traversal never saw is +// made entirely of orphan needles — the merge moves nothing, but the operator +// must be told the real cleanup is fsck. +func TestWarnUnreferencedSources(t *testing.T) { + vol := &master_pb.VolumeInformationMessage{Id: 203, FileCount: 18} + c := newMergeCmd(250000, vol) + plan := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{ + needle.VolumeId(203): {needle.VolumeId(187)}, + }} + + var sb strings.Builder + c.warnUnreferencedSources(&sb, plan, newSourceNeedleCounter(), "/buckets/test") + out := sb.String() + if !strings.Contains(out, "volume 203") || !strings.Contains(out, "orphan needles") { + t.Fatalf("expected orphan warning, got %q", out) + } + + // Needles seen during traversal: no warning. + seen := newSourceNeedleCounter() + for i := 0; i < 7; i++ { + seen.record(needle.VolumeId(203)) + } + sb.Reset() + c.warnUnreferencedSources(&sb, plan, seen, "/buckets/test") + if sb.Len() != 0 { + t.Fatalf("expected no warning when needles were seen, got %q", sb.String()) + } + + // Zero FileCount (truly empty volume): nothing to report. + emptyVol := &master_pb.VolumeInformationMessage{Id: 204} + c2 := newMergeCmd(250000, emptyVol) + plan2 := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{ + needle.VolumeId(204): {needle.VolumeId(187)}, + }} + sb.Reset() + c2.warnUnreferencedSources(&sb, plan2, newSourceNeedleCounter(), "/buckets/test") + if sb.Len() != 0 { + t.Fatalf("expected no warning for empty volume, got %q", sb.String()) + } +} + +// Concurrent BFS workers increment the production counter through record — the +// count must stay correct under -race with source-heavy inputs. +func TestWarnUnreferencedSources_ConcurrentRecording(t *testing.T) { + vol := &master_pb.VolumeInformationMessage{Id: 203, FileCount: 1000} + c := newMergeCmd(250000, vol) + plan := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{ + needle.VolumeId(203): {needle.VolumeId(187)}, + }} + + seen := newSourceNeedleCounter() + var wg sync.WaitGroup + for w := 0; w < 5; w++ { // TraverseBfs worker count + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 200; i++ { + seen.record(needle.VolumeId(203)) + } + }() + } + wg.Wait() + + if got := seen.count(needle.VolumeId(203)); got != 1000 { + t.Fatalf("lost increments under concurrency: %d", got) + } + + var sb strings.Builder + c.warnUnreferencedSources(&sb, plan, seen, "/buckets/test") + if sb.Len() != 0 { + t.Fatalf("expected no warning when all needles were seen, got %q", sb.String()) + } +}