diff --git a/weed/shell/command_fs_meta_save.go b/weed/shell/command_fs_meta_save.go index b936252f62..4034655350 100644 --- a/weed/shell/command_fs_meta_save.go +++ b/weed/shell/command_fs_meta_save.go @@ -323,6 +323,9 @@ func doTraverseBfsAndSaving(filerClient filer_pb.FilerClient, writer io.Writer, firstErr = genErr hasErr.Store(true) }) + close(outputChan) + wg.Wait() + <-saveErrChan return genErr } else { if e.IsDirectory { diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index 7578c71823..e3c301992d 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -271,19 +271,33 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. if *c.findMissingChunksInFiler { // collect all filer file ids and paths - if err = c.collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo, *purgeAbsent, collectModifyFromAtNs, collectCutoffFromAtNs); err != nil { - return fmt.Errorf("collectFilerFileIdAndPaths: %w", err) + var queuePurge func(purgeItem) + var finishPurge func() + if *applyPurging || *purgeAbsent { + queuePurge, finishPurge = startPurgeWorkers(*c.purgeConcurrency, c.httpDelete) } - for dataNodeId, volumeIdToVInfo := range dataNodeVolumeIdToVInfo { - // for each volume, check filer file ids - if err = c.findFilerChunksMissingInVolumeServers(volumeIdToVInfo, dataNodeId, *applyPurging || *purgeAbsent); err != nil { - return fmt.Errorf("findFilerChunksMissingInVolumeServers: %w", err) + err = func() error { + if err := c.collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo, *purgeAbsent, queuePurge, collectModifyFromAtNs, collectCutoffFromAtNs); err != nil { + return fmt.Errorf("collectFilerFileIdAndPaths: %w", err) } + for dataNodeId, volumeIdToVInfo := range dataNodeVolumeIdToVInfo { + // for each volume, check filer file ids + if err := c.findFilerChunksMissingInVolumeServers(volumeIdToVInfo, dataNodeId, queuePurge); err != nil { + return fmt.Errorf("findFilerChunksMissingInVolumeServers: %w", err) + } + } + return nil + }() + if finishPurge != nil { + finishPurge() + } + if err != nil { + return err } c.purgeEmptyDirectories() } else { // collect all filer file ids - if err = c.collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo, false, 0, 0); err != nil { + if err = c.collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo, false, nil, 0, 0); err != nil { return fmt.Errorf("failed to collect file ids from filer: %w", err) } // If any entry's manifest could not be resolved, our in-use fid set @@ -306,7 +320,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. return nil } -func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, purgeAbsent bool, collectModifyFromAtNs int64, cutoffFromAtNs int64) error { +func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, purgeAbsent bool, queuePurge func(purgeItem), collectModifyFromAtNs int64, cutoffFromAtNs int64) error { if *c.verbose { fmt.Fprintf(c.writer, "checking each file from filer path %s...\n", c.getCollectFilerFilePath()) } @@ -382,12 +396,6 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m }, func(outputChan chan interface{}) error { buffer := make([]byte, readbufferSize) - var queuePurge func(purgeItem) - var finishPurge func() - if purgeAbsent { - queuePurge, finishPurge = startPurgeWorkers(*c.purgeConcurrency, c.httpDelete) - defer finishPurge() - } for item := range outputChan { i := item.(*Item) if f, ok := files[i.vid]; ok { @@ -400,6 +408,13 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m if _, err := f.Write([]byte(i.path)); err != nil { return err } + util.Uint32toBytes(buffer[:4], uint32(len(i.hardLinkId))) + if _, err := f.Write(buffer[:4]); err != nil { + return err + } + if _, err := f.Write(i.hardLinkId); err != nil { + return err + } } else if *c.findMissingChunksInFiler { // check if the volume matches the filter if len(c.volumeIds) > 0 { @@ -423,10 +438,10 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m }) } -func (c *commandVolumeFsck) findFilerChunksMissingInVolumeServers(volumeIdToVInfo map[uint32]VInfo, dataNodeId string, applyPurging bool) error { +func (c *commandVolumeFsck) findFilerChunksMissingInVolumeServers(volumeIdToVInfo map[uint32]VInfo, dataNodeId string, queuePurge func(purgeItem)) error { for volumeId, vinfo := range volumeIdToVInfo { - checkErr := c.oneVolumeFileIdsCheckOneVolume(dataNodeId, volumeId, applyPurging) + checkErr := c.oneVolumeFileIdsCheckOneVolume(dataNodeId, volumeId, queuePurge) if checkErr != nil { return fmt.Errorf("failed to collect file ids from volume %d on %s: %v", volumeId, vinfo.server, checkErr) } @@ -650,7 +665,7 @@ func (lw *lockedWriter) Write(p []byte) (int, error) { return lw.w.Write(p) } -func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleId types.NeedleId, itemPath util.FullPath)) error { +func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleId types.NeedleId, itemPath util.FullPath, hardLinkId []byte)) error { fp, err := os.Open(getFilerFileIdFile(c.tempFolder, volumeId)) if err != nil { return err @@ -685,13 +700,23 @@ func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleI return fmt.Errorf("read fid path size mismatch for volume %d,%x%08x: got %d want %d", volumeId, item.fileKey, item.cookie, n, pathSize) } item.path = util.FullPath(pathBytes) + item.hardLinkId = nil + if _, err := io.ReadFull(br, buffer[:4]); err != nil { + return fmt.Errorf("read fid hard link id size for volume %d,%x%08x: %w", volumeId, item.fileKey, item.cookie, err) + } + if hardLinkIdSize := util.BytesToUint32(buffer[:4]); hardLinkIdSize > 0 { + item.hardLinkId = make([]byte, int(hardLinkIdSize)) + if _, err := io.ReadFull(br, item.hardLinkId); err != nil { + return fmt.Errorf("read fid hard link id for volume %d,%x%08x: %w", volumeId, item.fileKey, item.cookie, err) + } + } needleId := types.NeedleId(item.fileKey) - fn(needleId, item.path) + fn(needleId, item.path, item.hardLinkId) } return nil } -func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(dataNodeId string, volumeId uint32, applyPurging bool) (err error) { +func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(dataNodeId string, volumeId uint32, queuePurge func(purgeItem)) (err error) { if *c.verbose { fmt.Fprintf(c.writer, "find missing file chunks in dataNodeId %s volume %d ...\n", dataNodeId, volumeId) } @@ -702,11 +727,11 @@ func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(dataNodeId string, vo if err = db.LoadFromIdx(getVolumeFileIdFile(c.tempFolder, dataNodeId, volumeId)); err != nil { return } - if err = c.readFilerFileIdFile(volumeId, func(needleId types.NeedleId, itemPath util.FullPath) { + if err = c.readFilerFileIdFile(volumeId, func(needleId types.NeedleId, itemPath util.FullPath, hardLinkId []byte) { if _, found := db.Get(needleId); !found { fmt.Fprintf(c.writer, "%s\n", itemPath) - if applyPurging { - c.httpDelete(itemPath) + if queuePurge != nil { + queuePurge(purgeItem{path: itemPath, hardLinkId: hardLinkId}) } } }); err != nil { @@ -889,7 +914,7 @@ func (c *commandVolumeFsck) oneVolumeFileIdsSubtractFilerFileIds(dataNodeId stri return } - if err = c.readFilerFileIdFile(volumeId, func(filerNeedleId types.NeedleId, itemPath util.FullPath) { + if err = c.readFilerFileIdFile(volumeId, func(filerNeedleId types.NeedleId, itemPath util.FullPath, _ []byte) { inUseCount++ if *c.verifyNeedle && !vinfo.isEcVolume { if needleValue, ok := volumeFileIdDb.Get(filerNeedleId); ok && !needleValue.Size.IsDeleted() { diff --git a/weed/shell/command_volume_fsck_test.go b/weed/shell/command_volume_fsck_test.go index 50bdd0f8fa..7804b85224 100644 --- a/weed/shell/command_volume_fsck_test.go +++ b/weed/shell/command_volume_fsck_test.go @@ -1,12 +1,14 @@ package shell import ( + "bytes" "context" "fmt" "io" "net" "net/http" "net/http/httptest" + "os" "sort" "strings" "sync/atomic" @@ -16,6 +18,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle_map" + "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -310,3 +314,109 @@ func TestVolumeFsckPurgeHardLinksSerialized(t *testing.T) { t.Fatalf("processed %d deletes, want 8", got) } } + +// Orphans found by the per-volume check go through the same purge queue as +// the missing-volume path, and the fid file keeps each entry's hard link id +// so same-id deletes still land on one worker. +func TestVolumeFsckCheckOneVolumeQueuesOrphansWithHardLinkId(t *testing.T) { + tempDir := t.TempDir() + var verbose bool + var buf bytes.Buffer + c := &commandVolumeFsck{tempFolder: tempDir, verbose: &verbose, writer: &buf} + + volumeId := uint32(7) + db := needle_map.NewMemDb() + if err := db.Set(types.NeedleId(11), types.ToOffset(16), types.Size(1)); err != nil { + t.Fatalf("seed volume idx: %v", err) + } + if err := db.SaveToIdx(getVolumeFileIdFile(tempDir, "dn1", volumeId)); err != nil { + t.Fatalf("write volume idx: %v", err) + } + db.Close() + + f, err := os.Create(getFilerFileIdFile(tempDir, volumeId)) + if err != nil { + t.Fatalf("create fid file: %v", err) + } + writeFid := func(fileKey uint64, cookie uint32, p string, hardLinkId []byte) { + buffer := make([]byte, readbufferSize) + util.Uint64toBytes(buffer, fileKey) + util.Uint32toBytes(buffer[8:], cookie) + util.Uint32toBytes(buffer[12:], uint32(len(p))) + if _, err := f.Write(buffer); err != nil { + t.Fatalf("write fid header: %v", err) + } + if _, err := f.Write([]byte(p)); err != nil { + t.Fatalf("write fid path: %v", err) + } + util.Uint32toBytes(buffer[:4], uint32(len(hardLinkId))) + if _, err := f.Write(buffer[:4]); err != nil { + t.Fatalf("write fid hard link id size: %v", err) + } + if _, err := f.Write(hardLinkId); err != nil { + t.Fatalf("write fid hard link id: %v", err) + } + } + writeFid(11, 1, "/keep/a", nil) + writeFid(22, 2, "/orphan/b", []byte{1, 2, 3}) + writeFid(33, 3, "/orphan/c", nil) + if err := f.Close(); err != nil { + t.Fatalf("close fid file: %v", err) + } + + var queued []purgeItem + if err := c.oneVolumeFileIdsCheckOneVolume("dn1", volumeId, func(it purgeItem) { + queued = append(queued, it) + }); err != nil { + t.Fatalf("check volume: %v", err) + } + if len(queued) != 2 { + t.Fatalf("queued %d purges, want 2 (the two orphans): %v", len(queued), queued) + } + if queued[0].path != "/orphan/b" || queued[1].path != "/orphan/c" { + t.Fatalf("queued paths %v, want /orphan/b then /orphan/c", queued) + } + if !bytes.Equal(queued[0].hardLinkId, []byte{1, 2, 3}) { + t.Fatalf("queued hardLinkId %v, want [1 2 3]", queued[0].hardLinkId) + } + if queued[1].hardLinkId != nil { + t.Fatalf("queued hardLinkId %v, want nil", queued[1].hardLinkId) + } +} + +// A .fid file cut short inside the hard link id field is corrupt input, not a +// record without a link id: the reader must fail instead of emitting the entry +// with a nil id. +func TestReadFilerFileIdFileRejectsTruncatedHardLinkId(t *testing.T) { + tempDir := t.TempDir() + var verbose bool + c := &commandVolumeFsck{tempFolder: tempDir, verbose: &verbose, writer: io.Discard} + + f, err := os.Create(getFilerFileIdFile(tempDir, 7)) + if err != nil { + t.Fatalf("create fid file: %v", err) + } + buffer := make([]byte, readbufferSize) + util.Uint64toBytes(buffer, 11) + util.Uint32toBytes(buffer[8:], 1) + util.Uint32toBytes(buffer[12:], uint32(len("/orphan/b"))) + if _, err := f.Write(buffer); err != nil { + t.Fatalf("write fid header: %v", err) + } + if _, err := f.Write([]byte("/orphan/b")); err != nil { + t.Fatalf("write fid path: %v", err) + } + if _, err := f.Write([]byte{0, 0}); err != nil { + t.Fatalf("write partial hard link id size: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("close fid file: %v", err) + } + + err = c.readFilerFileIdFile(7, func(types.NeedleId, util.FullPath, []byte) { + t.Fatal("callback fired for a truncated record") + }) + if err == nil { + t.Fatal("expected an error for a truncated hard link id size") + } +} diff --git a/weed/storage/disk_location_ec_realworld_test.go b/weed/storage/disk_location_ec_realworld_test.go index 42ce29bd78..f54682a47d 100644 --- a/weed/storage/disk_location_ec_realworld_test.go +++ b/weed/storage/disk_location_ec_realworld_test.go @@ -5,9 +5,24 @@ import ( "path/filepath" "testing" + "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" ) +// requireDiskSpaceForEncoding skips the subtest when dir's filesystem cannot +// hold the .dat file plus the shards EC encoding writes for it, so ENOSPC on +// space-constrained runners doesn't surface as a test failure. +func requireDiskSpaceForEncoding(t *testing.T, dir string, datFileSize int64, expectedShardSize int64) { + t.Helper() + needBytes := uint64(datFileSize) + + uint64(expectedShardSize)*erasure_coding.TotalShardsCount + + 16*erasure_coding.ErasureCodingSmallBlockSize + if free := stats.NewDiskStatus(dir).Free; free < needBytes { + t.Skipf("insufficient free space in %s: have %d bytes, need ~%d to encode a %d-byte volume", + dir, free, needBytes, datFileSize) + } +} + // TestCalculateExpectedShardSizeWithRealEncoding validates our shard size calculation // by actually running EC encoding on real files and comparing the results func TestCalculateExpectedShardSizeWithRealEncoding(t *testing.T) { @@ -52,6 +67,10 @@ func TestCalculateExpectedShardSizeWithRealEncoding(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + // Calculate expected shard size using our function + expectedShardSize := calculateExpectedShardSize(tt.datFileSize, erasure_coding.DataShardsCount) + requireDiskSpaceForEncoding(t, tempDir, tt.datFileSize, expectedShardSize) + // Create a test .dat file with the specified size baseFileName := filepath.Join(tempDir, "test_volume") datFileName := baseFileName + ".dat" @@ -82,9 +101,6 @@ func TestCalculateExpectedShardSizeWithRealEncoding(t *testing.T) { } datFile.Close() - // Calculate expected shard size using our function - expectedShardSize := calculateExpectedShardSize(tt.datFileSize, erasure_coding.DataShardsCount) - // Run actual EC encoding _, err = erasure_coding.WriteEcFiles(baseFileName, erasure_coding.BackgroundECContext()) if err != nil { @@ -146,6 +162,9 @@ func TestCalculateExpectedShardSizeEdgeCases(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + expectedShardSize := calculateExpectedShardSize(tt.datFileSize, erasure_coding.DataShardsCount) + requireDiskSpaceForEncoding(t, tempDir, tt.datFileSize, expectedShardSize) + baseFileName := filepath.Join(tempDir, tt.name) datFileName := baseFileName + ".dat" @@ -163,9 +182,6 @@ func TestCalculateExpectedShardSizeEdgeCases(t *testing.T) { datFile.Write(data) datFile.Close() - // Calculate expected - expectedShardSize := calculateExpectedShardSize(tt.datFileSize, erasure_coding.DataShardsCount) - // Run actual EC encoding _, err = erasure_coding.WriteEcFiles(baseFileName, erasure_coding.BackgroundECContext()) if err != nil {