From 76198d15cfcc6dbbc9015f9f16fafcc5ff16bdd8 Mon Sep 17 00:00:00 2001 From: Minsoo Choi <162654709+Neo1228@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:57:43 +0900 Subject: [PATCH] fix(fsck): inspect references to deleted volumes (#12357) * fix(fsck): inspect references to deleted volumes * volume.fsck: treat every match-all collection filter as global ALL_COLLECTIONS and EACH_COLLECTION compile to the same match-all matcher as an empty filter or "*", but the string check treated them as a restricted scope: a selected missing volume was rejected and references to master-absent volumes were reported without being purged. Use the compiled matcher instead. Co-Authored-By: Claude Opus 5.5 * volume.fsck: purge an entry whose manifest is on a missing volume A top-level chunk on a missing volume already proves the entry is unreadable, whether or not its manifests resolve. Skipping such entries kept the very case where the manifest needle sat on the retired volume, and the remaining unresolved manifest then blocked every later orphan purge. The missing-needle check on live volumes never had this gate, so both paths now treat these entries the same. The warning now says only orphan purging is disabled, which is the only thing the counter gates. Co-Authored-By: Claude Opus 5.5 * volume.fsck: report a failed Filer delete A rejected purge DELETE printed nothing unless -v was set, so the run looked successful after "deleting path" while the entry stayed. The filer answers 204 for an entry that is already gone, so only real failures print. Co-Authored-By: Claude Opus 5.5 * volume.fsck: tighten help text, comments and missing-volume tests The help now wraps like the rest of it and says that a purge removes the whole entry, including its chunks on other volumes. The live test header no longer calls the run report-only, and the reverse-scan test checks the report line for each missing chunk. Co-Authored-By: Claude Opus 5.5 * volume.fsck: drop the missing-volume design notes The notes said no destructive change was approved while the change adds a confirmed purge, still described the dropped manifest gate, and mostly recorded local runs and a reverse index that is not part of this change. The operator contract lives in help volume.fsck. Co-Authored-By: Claude Opus 5.5 * volume.fsck: allow a report-only missing-volume scan with -collection The collection check exists to stop a purge that cannot be scoped, but it also rejected a report-only run that deletes nothing. Apply it only with -reallyDeleteFilerEntries, and limit the -confirmMissingVolumeDeletion description to the -volumeId case it guards. Co-Authored-By: Claude Opus 5.5 * volume.fsck: count only a 2xx answer as a purged entry A redirect the client could not follow left the entry in place but was recorded as deleted, which also queued its parent for empty-directory cleanup. Co-Authored-By: Claude Opus 5.5 * volume.fsck: say that a filtered purge of a missing -volumeId is refused The help and the early warning both said a purge with -collection only reports references to unlisted volumes, but a -volumeId naming one is rejected before the scan. The help now says so, and the warning prints after that check, so a refused run no longer claims it will report. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: MZ01-CHOIMINSU Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/shell/command_volume_fsck.go | 103 ++++- weed/shell/command_volume_fsck_live_test.go | 360 ++++++++++++++++++ ...command_volume_fsck_missing_volume_test.go | 184 +++++++++ weed/shell/command_volume_fsck_test.go | 6 +- 4 files changed, 632 insertions(+), 21 deletions(-) create mode 100644 weed/shell/command_volume_fsck_live_test.go create mode 100644 weed/shell/command_volume_fsck_missing_volume_test.go diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index 67b0fa08e5..efabf1ceb1 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -105,6 +105,16 @@ func (c *commandVolumeFsck) Help() string { walked directory by directory; there, avoid running with -reallyDeleteFromVolume while files are being renamed or moved. + With -findMissingChunksInFiler, -volumeId may name a volume the master no + longer lists, to find the filer entries that still reference it. A volume is + also unlisted while its server is offline, so purging these entries with + -reallyDeleteFilerEntries needs -confirmMissingVolumeDeletion as well; pass + it only after checking that every expected volume server is online. A purge + removes the whole filer entry, including its chunks on other volumes. + A -collection filter cannot be checked against an unlisted volume. With a + filter, a purge only reports references to unlisted volumes, and it refuses + to run when -volumeId names one. + ` } @@ -123,6 +133,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. applyPurging := fsckCommand.Bool("reallyDeleteFromVolume", false, " after detection, delete missing data from volumes / delete missing file entries from filer. Currently this only works with default filerGroup.") c.forcePurging = fsckCommand.Bool("forcePurging", false, "delete missing data from volumes in one replica used together with applyPurging") purgeAbsent := fsckCommand.Bool("reallyDeleteFilerEntries", false, " delete missing file entries from filer if the corresponding volume is missing for any reason, please ensure all still existing/expected volumes are connected! used together with findMissingChunksInFiler") + confirmMissingVolumeDeletion := fsckCommand.Bool("confirmMissingVolumeDeletion", false, "with -reallyDeleteFilerEntries, confirm that the -volumeId volumes the master does not list are retired") c.purgeConcurrency = fsckCommand.Int("purgeConcurrency", 16, "parallel filer deletes for -reallyDeleteFilerEntries") tempPath := fsckCommand.String("tempPath", path.Join(os.TempDir()), "path for temporary idx files") cutoffTimeAgo := fsckCommand.Duration("cutoffTimeAgo", 5*time.Hour, "only include entries on volume servers before this cutoff time to check orphan chunks") @@ -194,28 +205,26 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. } } - // If the operator specified -volumeId, reject unknown ids up front instead - // of silently filtering them out and reporting "no orphan data". A silent - // success on a missing volume looks identical to a clean volume and hides - // typos, already-deleted volumes, and stale scripts. if len(c.volumeIds) > 0 { - known := make(map[uint32]bool) - for _, vidMap := range dataNodeVolumeIdToVInfo { - for vid := range vidMap { - known[vid] = true - } - } - var missing []uint32 - for vid := range c.volumeIds { - if !known[vid] { - missing = append(missing, vid) - } + missing, err := validateSelectedVolumeIDs(c.volumeIds, c.knownVolumeIds, *c.findMissingChunksInFiler) + if err != nil { + return err } if len(missing) > 0 { - sort.Slice(missing, func(i, j int) bool { return missing[i] < missing[j] }) - return fmt.Errorf("volume(s) not found on master: %v", missing) + if *purgeAbsent { + if err := validateMissingVolumeCollectionFilter(missing, c.collectionMatcher); err != nil { + return err + } + } + if err := validateMissingVolumePurgeConfirmation(missing, *purgeAbsent, *confirmMissingVolumeDeletion); err != nil { + return err + } + fmt.Fprintf(c.writer, "WARNING: volume(s) not found on master (deleted or temporarily unavailable); scanning filer references for selected volume(s): %v\n", missing) } } + if *c.findMissingChunksInFiler && *purgeAbsent && c.collectionMatcher != nil { + fmt.Fprintln(c.writer, "WARNING: collection cannot be verified for volumes absent from master; Filer entries referencing those volumes will be reported but not purged.") + } var collectCutoffFromAtNs int64 = 0 if cutoffTimeAgo.Seconds() != 0 { @@ -360,13 +369,15 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m // unknown), warn, and keep going. The unresolved counter blocks // any purge step downstream so we never delete a sub-chunk we // couldn't account for. - fmt.Fprintf(c.writer, "WARNING: ResolveChunkManifest failed for %s: %v — recording top-level chunk fids only; purging will be disabled\n", + fmt.Fprintf(c.writer, "WARNING: ResolveChunkManifest failed for %s: %v — recording top-level chunk fids only; orphan purging will be disabled\n", util.NewFullPath(entry.Dir, entry.Entry.Name), resolveErr) c.unresolvedManifestEntries.Add(1) dataChunks = entry.Entry.GetChunks() manifestChunks = nil } dataChunks = append(dataChunks, manifestChunks...) + // One Filer entry can have many missing chunks; delete the path only once. + purgePathQueued := false for _, chunk := range dataChunks { if cutoffFromAtNs != 0 && chunk.ModifiedTsNs > cutoffFromAtNs { continue @@ -374,6 +385,17 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m if collectModifyFromAtNs != 0 && chunk.ModifiedTsNs < collectModifyFromAtNs { continue } + selectedVolume := len(c.volumeIds) == 0 + if len(c.volumeIds) > 0 { + _, selectedVolume = c.volumeIds[chunk.Fid.VolumeId] + } + purgePath := purgeAbsent && + selectedVolume && + !c.knownVolumeIds[chunk.Fid.VolumeId] && + !purgePathQueued && c.collectionMatcher == nil + if purgePath { + purgePathQueued = true + } select { case outputChan <- &Item{ vid: chunk.Fid.VolumeId, @@ -381,6 +403,7 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m cookie: chunk.Fid.Cookie, path: util.NewFullPath(entry.Dir, entry.Entry.Name), hardLinkId: entry.Entry.GetHardLinkId(), + purgePath: purgePath, }: case <-ctx.Done(): return ctx.Err() @@ -422,7 +445,7 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m continue } fmt.Fprintf(c.writer, "%d,%x%08x %s volume not found\n", i.vid, i.fileKey, i.cookie, i.path) - if purgeAbsent { + if purgeAbsent && i.purgePath { fmt.Fprintf(c.writer, "deleting path %s after volume not found\n", i.path) queuePurge(purgeItem{path: i.path, hardLinkId: i.hardLinkId}) } @@ -644,6 +667,7 @@ type Item struct { cookie uint32 path util.FullPath hardLinkId []byte + purgePath bool } // lockedWriter serializes concurrent writers — the traversal, the consumer, @@ -814,7 +838,9 @@ func (c *commandVolumeFsck) httpDelete(path util.FullPath) { fmt.Fprintln(c.writer, "delete response Headers : ", resp.Header) } - if resp.StatusCode < http.StatusBadRequest { + if resp.StatusCode >= http.StatusMultipleChoices { + fmt.Fprintf(c.writer, "DELETE %s failed: %s\n", path, resp.Status) + } else { dir, _ := path.DirAndName() c.purgedDirsLock.Lock() c.purgedDirs[util.FullPath(dir)] = struct{}{} @@ -1094,3 +1120,40 @@ func writeToFile(bytes []byte, fileName string) error { _, err = dst.Write(bytes) return err } + +// validateSelectedVolumeIDs rejects IDs the master does not list, unless the +// reverse filer scan is looking for references to exactly those volumes. +func validateSelectedVolumeIDs(requested, known map[uint32]bool, allowMissing bool) ([]uint32, error) { + var missing []uint32 + for vid := range requested { + if !known[vid] { + missing = append(missing, vid) + } + } + if len(missing) == 0 { + return nil, nil + } + sort.Slice(missing, func(i, j int) bool { return missing[i] < missing[j] }) + if !allowMissing { + return nil, fmt.Errorf("volume(s) not found on master: %v", missing) + } + return missing, nil +} + +// validateMissingVolumeCollectionFilter fails closed because a deleted volume +// has no master metadata from which fsck can verify its collection. +func validateMissingVolumeCollectionFilter(missing []uint32, matcher *wildcard.CollectionMatcher) error { + if len(missing) == 0 || matcher == nil { + return nil + } + return fmt.Errorf("cannot apply -collection=%q to missing volume(s) %v: their collection is unknown", matcher.String(), missing) +} + +// validateMissingVolumePurgeConfirmation requires operators to distinguish +// permanent retirement from a master that temporarily lost an offline server. +func validateMissingVolumePurgeConfirmation(missing []uint32, purgeAbsent, confirmed bool) error { + if len(missing) == 0 || !purgeAbsent || confirmed { + return nil + } + return fmt.Errorf("refusing to purge Filer entries for volume(s) absent from master: %v; verify all expected volumes are connected and pass -confirmMissingVolumeDeletion", missing) +} diff --git a/weed/shell/command_volume_fsck_live_test.go b/weed/shell/command_volume_fsck_live_test.go new file mode 100644 index 0000000000..15f35730ba --- /dev/null +++ b/weed/shell/command_volume_fsck_live_test.go @@ -0,0 +1,360 @@ +//go:build integration + +package shell + +import ( + "bytes" + "context" + "fmt" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "io" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "regexp" + "runtime" + "strconv" + "strings" + "testing" + "time" +) + +// Opt-in: runs report-only and confirmed-purge scans against real services +// after a volume is deleted, and checks that a stopped server blocks the purge. +func TestVolumeFsckFindsReferenceAfterRealVolumeDeletion(t *testing.T) { + weed := buildWeedForIntegration(t) + base := t.TempDir() + for _, name := range []string{"logs", "master", "volume", "filer"} { + if err := os.MkdirAll(filepath.Join(base, name), 0o755); err != nil { + t.Fatal(err) + } + } + + portPairs := freeIntegrationPortPairs(t, 3) + masterHTTP, masterGRPC := portPairs[0][0], portPairs[0][1] + volumeHTTP, volumeGRPC := portPairs[1][0], portPairs[1][1] + filerHTTP, filerGRPC := portPairs[2][0], portPairs[2][1] + masterAddress := fmt.Sprintf("127.0.0.1:%d", masterHTTP) + + startIntegrationProcess(t, base, weed, "master", + "master", "-ip=127.0.0.1", "-ip.bind=127.0.0.1", + fmt.Sprintf("-port=%d", masterHTTP), fmt.Sprintf("-port.grpc=%d", masterGRPC), + "-mdir="+filepath.Join(base, "master"), "-defaultReplication=000", "-volumeSizeLimitMB=64") + waitIntegrationTCP(t, masterHTTP, filepath.Join(base, "logs", "master.log")) + waitIntegrationTCP(t, masterGRPC, filepath.Join(base, "logs", "master.log")) + + volumeProcess := startIntegrationProcess(t, base, weed, "volume", + "volume", "-ip=127.0.0.1", "-ip.bind=127.0.0.1", + fmt.Sprintf("-port=%d", volumeHTTP), fmt.Sprintf("-port.grpc=%d", volumeGRPC), + "-master="+masterAddress, "-dir="+filepath.Join(base, "volume"), "-max=10") + waitIntegrationTCP(t, volumeHTTP, filepath.Join(base, "logs", "volume.log")) + waitIntegrationTCP(t, volumeGRPC, filepath.Join(base, "logs", "volume.log")) + waitForMasterVolumeRegistration(t, masterHTTP, volumeHTTP, filepath.Join(base, "logs", "volume.log")) + + startIntegrationProcess(t, base, weed, "filer", + "filer", "-ip=127.0.0.1", "-ip.bind=127.0.0.1", + fmt.Sprintf("-port=%d", filerHTTP), fmt.Sprintf("-port.grpc=%d", filerGRPC), + "-master="+masterAddress, "-defaultReplicaPlacement=000", + "-defaultStoreDir="+filepath.Join(base, "filer")) + waitIntegrationTCP(t, filerHTTP, filepath.Join(base, "logs", "filer.log")) + waitIntegrationTCP(t, filerGRPC, filepath.Join(base, "logs", "filer.log")) + + filerURL := fmt.Sprintf("http://127.0.0.1:%d", filerHTTP) + request, err := http.NewRequest(http.MethodPut, filerURL+"/fsck-probe.txt", strings.NewReader("live integration data")) + if err != nil { + t.Fatal(err) + } + response, err := http.DefaultClient.Do(request) + if err != nil { + t.Fatalf("upload probe through Filer: %v", err) + } + response.Body.Close() + if response.StatusCode != http.StatusCreated && response.StatusCode != http.StatusAccepted { + t.Fatalf("Filer PUT status = %d, want 201 or 202", response.StatusCode) + } + + filerAddress := fmt.Sprintf("127.0.0.1:%d", filerHTTP) + meta := runIntegrationShell(t, weed, masterAddress, filerAddress, "lock\nfs.meta.cat /fsck-probe.txt\nunlock\n") + fidPattern := regexp.MustCompile(`"fid"\s*:\s*\{\s*"volumeId"\s*:\s*(\d+)`) + match := fidPattern.FindStringSubmatch(meta) + if len(match) != 2 { + t.Fatalf("could not read chunk volume ID from Filer metadata: %s", meta) + } + volumeID := match[1] + + // Delete only the physical volume. The Filer entry intentionally remains + // so fsck must discover a dangling reference rather than a missing file. + deleteOutput := runIntegrationShell(t, weed, masterAddress, filerAddress, + fmt.Sprintf("lock\nvolume.delete -node=127.0.0.1:%d -volumeId=%s\nunlock\n", volumeHTTP, volumeID)) + recentOutput := runIntegrationShell(t, weed, masterAddress, filerAddress, + fmt.Sprintf("lock\nvolume.fsck -findMissingChunksInFiler -volumeId=%s\nunlock\n", volumeID)) + if strings.Contains(recentOutput, "/fsck-probe.txt") { + t.Fatalf("default cutoff unexpectedly reported a just-written file: %q", recentOutput) + } + + fsckOutput := runIntegrationShell(t, weed, masterAddress, filerAddress, + fmt.Sprintf("lock\nvolume.fsck -findMissingChunksInFiler -volumeId=%s -cutoffTimeAgo=0\nunlock\n", volumeID)) + if !strings.Contains(fsckOutput, "scanning filer references for selected volume(s)") { + t.Fatalf("fsck did not treat the deleted ID as a missing-volume target; delete=%q fsck=%q", deleteOutput, fsckOutput) + } + if !strings.Contains(fsckOutput, "/fsck-probe.txt") || !strings.Contains(fsckOutput, "volume not found") { + t.Fatalf("fsck did not report the stale Filer reference: %q", fsckOutput) + } + + scopedPurgeOutput := runIntegrationShell(t, weed, masterAddress, filerAddress, + "lock\nvolume.fsck -findMissingChunksInFiler -reallyDeleteFilerEntries -collection=logs -cutoffTimeAgo=0\nunlock\n") + if !strings.Contains(scopedPurgeOutput, "collection cannot be verified for volumes absent from master") { + t.Fatalf("fsck did not warn that a missing volume cannot be collection-scoped: %q", scopedPurgeOutput) + } + if strings.Contains(scopedPurgeOutput, "deleting path /fsck-probe.txt after volume not found") { + t.Fatalf("collection-filtered fsck queued an unverified missing-volume path: %q", scopedPurgeOutput) + } + + // Report-only fsck must not delete the stale Filer entry. + metaAfter := runIntegrationShell(t, weed, masterAddress, filerAddress, "lock\nfs.meta.cat /fsck-probe.txt\nunlock\n") + if !strings.Contains(metaAfter, "fsck-probe.txt") { + t.Fatalf("report-only fsck changed or removed the Filer entry: %q", metaAfter) + } + // An entry whose chunk manifest lived on the retired volume is unreadable too. + filerConn, err := grpc.NewClient(fmt.Sprintf("127.0.0.1:%d", filerGRPC), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatalf("dial Filer gRPC: %v", err) + } + defer filerConn.Close() + filerClient := filer_pb.NewSeaweedFilerClient(filerConn) + volumeIDUint, err := strconv.ParseUint(volumeID, 10, 32) + if err != nil { + t.Fatalf("parse volume ID %q: %v", volumeID, err) + } + manifestChunk := &filer_pb.FileChunk{ + Fid: &filer_pb.FileId{VolumeId: uint32(volumeIDUint), FileKey: 987654321, Cookie: 1}, + Size: 21, + IsChunkManifest: true, + } + manifestChunk.FileId = manifestChunk.GetFileIdString() + now := time.Now().Unix() + if err := filer_pb.CreateEntry(context.Background(), filerClient, &filer_pb.CreateEntryRequest{ + Directory: "/", + Entry: &filer_pb.Entry{ + Name: "manifest-probe.txt", + Attributes: &filer_pb.FuseAttributes{FileSize: 21, FileMode: 0o644, Mtime: now, Crtime: now}, + Chunks: []*filer_pb.FileChunk{manifestChunk}, + }, + }); err != nil { + t.Fatalf("create unresolved-manifest probe: %v", err) + } + + purgeOutput := runIntegrationShell(t, weed, masterAddress, filerAddress, + fmt.Sprintf("lock\nvolume.fsck -findMissingChunksInFiler -reallyDeleteFilerEntries -confirmMissingVolumeDeletion -volumeId=%s -cutoffTimeAgo=0\nunlock\n", volumeID)) + if !strings.Contains(purgeOutput, "ResolveChunkManifest failed") { + t.Fatalf("fsck did not surface the unresolved manifest: %q", purgeOutput) + } + if !strings.Contains(purgeOutput, "deleting path /manifest-probe.txt") { + t.Fatalf("confirmed purge did not queue the entry whose manifest was on the retired volume: %q", purgeOutput) + } + manifestAfterPurge, err := http.Get(filerURL + "/manifest-probe.txt") + if err != nil { + t.Fatalf("read manifest probe after confirmed purge: %v", err) + } + manifestAfterPurge.Body.Close() + if manifestAfterPurge.StatusCode != http.StatusNotFound { + t.Fatalf("confirmed purge manifest probe GET status = %d, want 404", manifestAfterPurge.StatusCode) + } + + if !strings.Contains(purgeOutput, "deleting path /fsck-probe.txt after volume not found") { + t.Fatalf("confirmed purge did not queue the direct missing-volume entry: %q", purgeOutput) + } + probeAfterPurge, err := http.Get(filerURL + "/fsck-probe.txt") + if err != nil { + t.Fatalf("read Filer probe after confirmed purge: %v", err) + } + probeAfterPurge.Body.Close() + if probeAfterPurge.StatusCode != http.StatusNotFound { + t.Fatalf("confirmed purge Filer GET status = %d, want 404", probeAfterPurge.StatusCode) + } + // A volume still listed by the master but whose server is unreachable is + // not proof that its references are stale. Purge must fail before touching + // the Filer entry when the volume index cannot be read. + offlineRequest, err := http.NewRequest(http.MethodPut, filerURL+"/offline-probe.txt", strings.NewReader("offline safety probe")) + if err != nil { + t.Fatal(err) + } + offlineResponse, err := http.DefaultClient.Do(offlineRequest) + if err != nil { + t.Fatalf("upload offline safety probe: %v", err) + } + offlineResponse.Body.Close() + if offlineResponse.StatusCode != http.StatusCreated && offlineResponse.StatusCode != http.StatusAccepted { + t.Fatalf("offline safety PUT status = %d", offlineResponse.StatusCode) + } + offlineMeta := runIntegrationShell(t, weed, masterAddress, filerAddress, "lock\nfs.meta.cat /offline-probe.txt\nunlock\n") + offlineMatch := fidPattern.FindStringSubmatch(offlineMeta) + if len(offlineMatch) != 2 { + t.Fatalf("could not read offline probe volume ID: %s", offlineMeta) + } + if err := volumeProcess.Process.Kill(); err != nil { + t.Fatalf("stop volume server: %v", err) + } + waitIntegrationTCPClosed(t, volumeHTTP) + + shellCtx, cancelShell := context.WithTimeout(context.Background(), 30*time.Second) + defer cancelShell() + fsckCmd := exec.CommandContext(shellCtx, weed, "shell", "-master="+masterAddress, "-filer="+filerAddress) + fsckCmd.Stdin = strings.NewReader(fmt.Sprintf("lock\nvolume.fsck -findMissingChunksInFiler -reallyDeleteFilerEntries -volumeId=%s -cutoffTimeAgo=0\nunlock\n", offlineMatch[1])) + fsckBytes, fsckErr := fsckCmd.CombinedOutput() + offlineFsckOutput := string(fsckBytes) + if fsckErr == nil { + t.Fatalf("fsck unexpectedly purged metadata for a volume server outage: %q", offlineFsckOutput) + } + if !strings.Contains(offlineFsckOutput, "confirmMissingVolumeDeletion") && !strings.Contains(offlineFsckOutput, "failed to collect file ids") { + t.Fatalf("fsck failed for an unexpected reason: err=%v output=%q", fsckErr, offlineFsckOutput) + } + offlineMetaAfter := runIntegrationShell(t, weed, masterAddress, filerAddress, "lock\nfs.meta.cat /offline-probe.txt\nunlock\n") + if !strings.Contains(offlineMetaAfter, "offline-probe.txt") { + t.Fatalf("offline-volume fsck removed the live Filer entry: %q", offlineMetaAfter) + } +} + +func buildWeedForIntegration(t *testing.T) string { + t.Helper() + // Reuse the CI/integration binary when supplied; otherwise build this checkout. + if binary := os.Getenv("WEED_BINARY"); binary != "" { + if _, err := os.Stat(binary); err != nil { + t.Fatalf("WEED_BINARY=%q is unavailable: %v", binary, err) + } + return binary + } + _, source, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("resolve integration test source path") + } + root := filepath.Clean(filepath.Join(filepath.Dir(source), "..", "..")) + binary := filepath.Join(t.TempDir(), "weed") + cmd := exec.Command("go", "build", "-o", binary, "./weed") + cmd.Dir = root + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("build weed: %v\n%s", err, out) + } + return binary +} + +// freeIntegrationPortPairs reserves HTTP ports with the CLI's conventional +// HTTP+10000 gRPC mapping. The low range also keeps the derived ports valid +// for weed shell's master/Filer client configuration. +func freeIntegrationPortPairs(t *testing.T, count int) [][2]int { + t.Helper() + var listeners []net.Listener + defer func() { + for _, listener := range listeners { + listener.Close() + } + }() + + pairs := make([][2]int, 0, count) + for httpPort := 20000; httpPort < 30000 && len(pairs) < count; httpPort++ { + grpcPort := httpPort + 10000 + httpListener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", httpPort)) + if err != nil { + continue + } + grpcListener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", grpcPort)) + if err != nil { + httpListener.Close() + continue + } + listeners = append(listeners, httpListener, grpcListener) + pairs = append(pairs, [2]int{httpPort, grpcPort}) + } + if len(pairs) != count { + t.Fatalf("reserved %d of %d required HTTP/gRPC port pairs", len(pairs), count) + } + return pairs +} + +func startIntegrationProcess(t *testing.T, base, weed, name string, args ...string) *exec.Cmd { + t.Helper() + logFile, err := os.Create(filepath.Join(base, "logs", name+".log")) + if err != nil { + t.Fatal(err) + } + cmd := exec.Command(weed, args...) + cmd.Stdout = logFile + cmd.Stderr = logFile + if err := cmd.Start(); err != nil { + logFile.Close() + t.Fatalf("start %s: %v", name, err) + } + t.Cleanup(func() { + if cmd.Process != nil { + _ = cmd.Process.Kill() + } + _ = cmd.Wait() + _ = logFile.Close() + }) + return cmd +} + +// waitForMasterVolumeRegistration prevents racing a successful HTTP listener +// start against the volume server's first heartbeat to the master. +func waitForMasterVolumeRegistration(t *testing.T, masterPort, volumePort int, logPath string) { + t.Helper() + deadline := time.Now().Add(20 * time.Second) + wantAddress := fmt.Sprintf("127.0.0.1:%d", volumePort) + for time.Now().Before(deadline) { + response, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/dir/status", masterPort)) + if err == nil { + body, readErr := io.ReadAll(response.Body) + response.Body.Close() + if readErr == nil && bytes.Contains(body, []byte(wantAddress)) { + return + } + } + time.Sleep(100 * time.Millisecond) + } + log, _ := os.ReadFile(logPath) + t.Fatalf("master did not register volume server %s; log:\n%s", wantAddress, bytes.TrimSpace(log)) +} + +func waitIntegrationTCPClosed(t *testing.T, port int) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), 100*time.Millisecond) + if err != nil { + return + } + conn.Close() + time.Sleep(50 * time.Millisecond) + } + t.Fatalf("service still listens on port %d after process was stopped", port) +} + +func waitIntegrationTCP(t *testing.T, port int, logPath string) { + t.Helper() + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), 200*time.Millisecond) + if err == nil { + conn.Close() + return + } + time.Sleep(100 * time.Millisecond) + } + log, _ := os.ReadFile(logPath) + t.Fatalf("service did not listen on port %d; log:\n%s", port, bytes.TrimSpace(log)) +} + +func runIntegrationShell(t *testing.T, weed, master, filer, input string) string { + t.Helper() + cmd := exec.Command(weed, "shell", "-master="+master, "-filer="+filer) + cmd.Stdin = strings.NewReader(input) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("weed shell failed: %v\n%s", err, out) + } + return string(out) +} diff --git a/weed/shell/command_volume_fsck_missing_volume_test.go b/weed/shell/command_volume_fsck_missing_volume_test.go new file mode 100644 index 0000000000..f19e28aa30 --- /dev/null +++ b/weed/shell/command_volume_fsck_missing_volume_test.go @@ -0,0 +1,184 @@ +package shell + +import ( + "bytes" + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" +) + +// A selected volume can already be absent from the master when fsck is used to +// clean stale Filer references after physical volume deletion. Only that +// explicitly selected missing ID should enter the purge queue. +func TestVolumeFsckFindMissingFilerEntriesForDeletedSelectedVolume(t *testing.T) { + tempDir := t.TempDir() + chunk := func(volumeID uint32, fileKey uint64) *filer_pb.FileChunk { + return &filer_pb.FileChunk{ + Fid: &filer_pb.FileId{VolumeId: volumeID, FileKey: fileKey, Cookie: 1}, + Size: 1, + } + } + stub := &stubFilerServer{entries: map[string]*filer_pb.Entry{ + "/": {IsDirectory: true}, + "/archive": {IsDirectory: true}, + "/archive/old": {Chunks: []*filer_pb.FileChunk{chunk(99, 1), chunk(99, 4), chunk(100, 2), chunk(7, 5)}}, + "/live": {Chunks: []*filer_pb.FileChunk{chunk(7, 3)}}, + }} + + verbose := false + findMissing := true + var output bytes.Buffer + c := &commandVolumeFsck{ + tempFolder: tempDir, + verbose: &verbose, + findMissingChunksInFiler: &findMissing, + writer: &output, + env: startStubFiler(t, stub), + volumeIds: map[uint32]bool{99: true}, + knownVolumeIds: map[uint32]bool{7: true}, + } + currentVolumes := map[string]map[uint32]VInfo{"dn1": {7: {collection: "live"}}} + var purged []purgeItem + if err := c.collectFilerFileIdAndPaths(currentVolumes, true, func(item purgeItem) { + purged = append(purged, item) + }, 0, 0); err != nil { + t.Fatalf("collect filer references: %v", err) + } + + if len(purged) != 1 || purged[0].path != util.FullPath("/archive/old") { + t.Fatalf("purged paths = %v, want only /archive/old for selected missing volume 99", purged) + } + // The same logical file also references a live volume. The scan preserves + // that reference while queuing the path only once for the missing chunks. + foundSurvivingChunk := false + if err := c.readFilerFileIdFile(7, func(_ types.NeedleId, itemPath util.FullPath, _ []byte) { + if itemPath == util.FullPath("/archive/old") { + foundSurvivingChunk = true + } + }); err != nil { + t.Fatalf("read surviving volume references: %v", err) + } + if !foundSurvivingChunk { + t.Fatal("the surviving chunk reference was not recorded for volume 7") + } + for _, line := range []string{ + "99,100000001 /archive/old volume not found\n", + "99,400000001 /archive/old volume not found\n", + } { + if !strings.Contains(output.String(), line) { + t.Errorf("missing-volume report %q lacks %q", output.String(), line) + } + } + if strings.Contains(output.String(), "100,200000001") { + t.Errorf("missing-volume report %q includes unselected volume 100", output.String()) + } +} + +// Unknown IDs remain errors outside the reverse Filer scan, preserving typo detection. +func TestVolumeFsckSelectedVolumeIDValidation(t *testing.T) { + known := map[uint32]bool{7: true} + requested := map[uint32]bool{99: true, 100: true} + + if _, err := validateSelectedVolumeIDs(requested, known, false); err == nil { + t.Fatal("normal fsck accepted volume IDs not present on the master") + } + missing, err := validateSelectedVolumeIDs(requested, known, true) + if err != nil { + t.Fatalf("reverse Filer scan rejected missing volume IDs: %v", err) + } + if len(missing) != 2 || missing[0] != 99 || missing[1] != 100 { + t.Fatalf("missing IDs = %v, want sorted [99 100]", missing) + } +} + +// A missing volume has no trustworthy collection metadata; a collection +// restriction must fail closed instead of silently widening or narrowing scope. +func TestValidateMissingVolumeCollectionFilter(t *testing.T) { + missing := []uint32{12} + compile := func(filter string) *wildcard.CollectionMatcher { + matcher, err := wildcard.CompileCollectionMatcher(filter) + if err != nil { + t.Fatalf("compile %q: %v", filter, err) + } + return matcher + } + for _, filter := range []string{"", "*", "ALL_COLLECTIONS", "EACH_COLLECTION"} { + if err := validateMissingVolumeCollectionFilter(missing, compile(filter)); err != nil { + t.Errorf("filter %q: %v", filter, err) + } + } + if err := validateMissingVolumeCollectionFilter(missing, compile("photos")); err == nil || !strings.Contains(err.Error(), "collection is unknown") { + t.Fatalf("expected an unknown-collection error, got %v", err) + } +} + +// Missing IDs require an extra acknowledgement only when metadata deletion +// is requested; read-only diagnosis and known-volume purges are unchanged. +func TestValidateMissingVolumePurgeConfirmation(t *testing.T) { + missing := []uint32{12} + for _, test := range []struct { + name string + missing []uint32 + purgeAbsent bool + confirmed bool + wantError bool + }{ + {name: "read-only scan", missing: missing}, + {name: "purge known volumes only", purgeAbsent: true}, + {name: "confirmed missing volume purge", missing: missing, purgeAbsent: true, confirmed: true}, + {name: "unconfirmed missing volume purge", missing: missing, purgeAbsent: true, wantError: true}, + } { + t.Run(test.name, func(t *testing.T) { + err := validateMissingVolumePurgeConfirmation(test.missing, test.purgeAbsent, test.confirmed) + if (err != nil) != test.wantError { + t.Fatalf("validateMissingVolumePurgeConfirmation() error = %v, wantError %v", err, test.wantError) + } + if err != nil && !strings.Contains(err.Error(), "confirmMissingVolumeDeletion") { + t.Fatalf("error %q does not name the required acknowledgement flag", err) + } + }) + } +} + +func TestVolumeFsckPurgesEntryWhoseManifestIsOnMissingVolume(t *testing.T) { + manifest := &filer_pb.FileChunk{ + Fid: &filer_pb.FileId{VolumeId: 99, FileKey: 1, Cookie: 1}, + Size: 1, + IsChunkManifest: true, + } + stub := &stubFilerServer{entries: map[string]*filer_pb.Entry{ + "/": {IsDirectory: true}, + "/big": {Chunks: []*filer_pb.FileChunk{manifest}}, + "/live": {Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 7, FileKey: 2, Cookie: 1}, Size: 1}}}, + }} + + verbose := false + findMissing := true + var output bytes.Buffer + c := &commandVolumeFsck{ + tempFolder: t.TempDir(), + verbose: &verbose, + findMissingChunksInFiler: &findMissing, + writer: &output, + env: startStubFiler(t, stub), + volumeIds: map[uint32]bool{99: true}, + knownVolumeIds: map[uint32]bool{7: true}, + } + var purged []purgeItem + if err := c.collectFilerFileIdAndPaths(map[string]map[uint32]VInfo{"dn1": {7: {}}}, true, func(item purgeItem) { + purged = append(purged, item) + }, 0, 0); err != nil { + t.Fatalf("collect filer references: %v", err) + } + + if !strings.Contains(output.String(), "ResolveChunkManifest failed for /big") { + t.Fatalf("manifest on the missing volume resolved unexpectedly: %q", output.String()) + } + if len(purged) != 1 || purged[0].path != util.FullPath("/big") { + t.Fatalf("purged paths = %v, want /big, whose manifest is on missing volume 99", purged) + } +} diff --git a/weed/shell/command_volume_fsck_test.go b/weed/shell/command_volume_fsck_test.go index 0c742f82ab..5993688436 100644 --- a/weed/shell/command_volume_fsck_test.go +++ b/weed/shell/command_volume_fsck_test.go @@ -56,9 +56,10 @@ func TestVolumeFsckHttpDeleteRecordsParentDirectory(t *testing.T) { defer filer.Close() verbose := false + var output bytes.Buffer c := &commandVolumeFsck{ verbose: &verbose, - writer: io.Discard, + writer: &output, purgedDirs: make(map[util.FullPath]struct{}), env: &CommandEnv{option: &ShellOptions{FilerAddress: pb.ServerAddress(strings.TrimPrefix(filer.URL, "http://"))}}, } @@ -72,6 +73,9 @@ func TestVolumeFsckHttpDeleteRecordsParentDirectory(t *testing.T) { if _, found := c.purgedDirs["/locked/dir"]; found { t.Errorf("expected a rejected delete not to be recorded, got %v", c.purgedDirs) } + if got := output.String(); got != "DELETE /locked/dir/file.txt failed: 403 Forbidden\n" { + t.Errorf("delete output = %q, want only the rejected path", got) + } } // stubFilerServer keeps a flat set of full paths and enforces the same delete