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