diff --git a/.github/workflows/fuse-p2p-integration.yml b/.github/workflows/fuse-p2p-integration.yml new file mode 100644 index 000000000..8ab1e44b2 --- /dev/null +++ b/.github/workflows/fuse-p2p-integration.yml @@ -0,0 +1,69 @@ +name: "FUSE P2P Peer Chunk Sharing Integration Tests" + +on: + pull_request: + paths: + - 'weed/command/mount*.go' + - 'weed/mount/**' + - 'weed/filer/mount_peer_registry*.go' + - 'weed/server/filer_grpc_server_mount_peer.go' + - 'weed/pb/mount_peer.proto' + - 'weed/pb/filer.proto' + - 'test/fuse_p2p/**' + - '.github/workflows/fuse-p2p-integration.yml' + push: + branches: [master] + paths: + - 'weed/command/mount*.go' + - 'weed/mount/**' + - 'weed/filer/mount_peer_registry*.go' + - 'weed/server/filer_grpc_server_mount_peer.go' + - 'weed/pb/mount_peer.proto' + - 'weed/pb/filer.proto' + - 'test/fuse_p2p/**' + +concurrency: + group: ${{ github.head_ref || github.ref }}/fuse-p2p-integration + cancel-in-progress: true + +permissions: + contents: read + +jobs: + fuse-p2p-integration: + name: FUSE P2P Peer Chunk Sharing + runs-on: ubuntu-22.04 + timeout-minutes: 20 + + steps: + - name: Check out code + uses: actions/checkout@v6 + + - name: Set up Go + uses: actions/setup-go@v6 + with: + go-version-file: 'go.mod' + + - name: Install FUSE dependencies + run: | + sudo apt-get update + sudo apt-get install -y libfuse3-dev + echo 'user_allow_other' | sudo tee -a /etc/fuse.conf + sudo chmod 644 /etc/fuse.conf + + - name: Build SeaweedFS + run: go build -o weed/weed -buildvcs=false ./weed + + - name: Run P2P integration tests + timeout-minutes: 15 + env: + WEED_BINARY: ${{ github.workspace }}/weed/weed + run: go test -v -count=1 -timeout=12m ./test/fuse_p2p/... + + - name: Upload logs on failure + if: failure() + uses: actions/upload-artifact@v7 + with: + name: fuse-p2p-test-logs + path: /tmp/seaweedfs-fuse-p2p-logs/ + retention-days: 3 diff --git a/test/fuse_p2p/framework_test.go b/test/fuse_p2p/framework_test.go new file mode 100644 index 000000000..96287c043 --- /dev/null +++ b/test/fuse_p2p/framework_test.go @@ -0,0 +1,390 @@ +//go:build linux || darwin + +package fuse_p2p + +import ( + "fmt" + "net" + "os" + "os/exec" + "path/filepath" + "strconv" + "sync" + "syscall" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/testutil" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/stretchr/testify/require" +) + +// p2pTestCluster manages a minimal SeaweedFS cluster exercising the peer +// chunk-sharing (p2p) read path: 1 master, 1 volume, 1 filer, and N FUSE +// mounts that all have -peer.enable set. Three mounts is the sweet spot +// for integration testing — with 2 mounts the HRW owner of a chunk is +// ~50% the reader itself (which short-circuits the peer path); with 3+ +// mounts it's ≤ 1/3, so a multi-chunk file almost certainly exercises +// the remote-owner fan-out. +type p2pTestCluster struct { + t testing.TB + baseDir string + weedBinary string + + masterPort int + masterGrpcPort int + volumePort int + volumeGrpcPort int + filerPort int + filerGrpcPort int + mountPeerPorts []int + mountPoints []string + + masterCmd *exec.Cmd + volumeCmd *exec.Cmd + filerCmd *exec.Cmd + mountCmds []*exec.Cmd + logFiles []*os.File + + cleanupOnce sync.Once +} + +// startP2PTestCluster brings up a cluster with numMounts FUSE mounts, +// every one advertising on its own -peer.listen port. All mounts register +// with the same filer. Verbose logging (-v=4) is enabled on each mount so +// peer-read success/failure messages land in the log file and the test +// can grep them to verify the p2p path fired. +func startP2PTestCluster(t testing.TB, numMounts int) *p2pTestCluster { + require.GreaterOrEqual(t, numMounts, 2, "need at least 2 mounts to exercise p2p") + binary := findWeedBinary() + if binary == "" { + t.Skip("weed binary not found; set WEED_BINARY or ensure it is on PATH") + } + baseDir, err := os.MkdirTemp("", "seaweedfs_fuse_p2p_test_") + require.NoError(t, err) + + c := &p2pTestCluster{ + t: t, + baseDir: baseDir, + weedBinary: binary, + mountPeerPorts: make([]int, numMounts), + mountPoints: make([]string, numMounts), + mountCmds: make([]*exec.Cmd, numMounts), + } + t.Cleanup(c.Stop) + + // master(2) + volume(2) + filer(2) + one peer port per mount. + // testutil.AllocatePorts holds all listeners open until every port + // is reserved, avoiding the brief close→bind race that would + // happen with a per-listener close loop. + ports, err := testutil.AllocatePorts(6 + numMounts) + require.NoError(t, err) + c.masterPort = ports[0] + c.masterGrpcPort = ports[1] + c.volumePort = ports[2] + c.volumeGrpcPort = ports[3] + c.filerPort = ports[4] + c.filerGrpcPort = ports[5] + for i := 0; i < numMounts; i++ { + c.mountPeerPorts[i] = ports[6+i] + } + + configDir := filepath.Join(baseDir, "config") + require.NoError(t, os.MkdirAll(configDir, 0755)) + require.NoError(t, os.WriteFile(filepath.Join(configDir, "security.toml"), []byte(""), 0644)) + + require.NoError(t, c.startMaster(configDir)) + require.NoError(t, c.waitForTCP(c.masterCmd, "master", + fmt.Sprintf("127.0.0.1:%d", c.masterPort), 30*time.Second)) + + require.NoError(t, c.startVolume(configDir)) + require.NoError(t, c.waitForTCP(c.volumeCmd, "volume", + fmt.Sprintf("127.0.0.1:%d", c.volumePort), 30*time.Second)) + + require.NoError(t, c.startFiler(configDir)) + require.NoError(t, c.waitForTCP(c.filerCmd, "filer", + fmt.Sprintf("127.0.0.1:%d", c.filerGrpcPort), 30*time.Second)) + + for i := 0; i < numMounts; i++ { + mp := filepath.Join(baseDir, fmt.Sprintf("mount%d", i)) + require.NoError(t, os.MkdirAll(mp, 0755)) + c.mountPoints[i] = mp + require.NoError(t, c.startMount(i, configDir)) + require.NoError(t, c.waitForMount(mp, 30*time.Second), + "mount %d not ready\n%s", i, c.tailLog(fmt.Sprintf("mount%d", i))) + } + return c +} + +func (c *p2pTestCluster) Stop() { + if c == nil { + return + } + c.cleanupOnce.Do(func() { + for i := len(c.mountCmds) - 1; i >= 0; i-- { + stopCmd(c.mountCmds[i]) + // Backup unmount in case the FUSE teardown didn't clean up. + exec.Command("fusermount3", "-u", c.mountPoints[i]).Run() + exec.Command("fusermount", "-u", c.mountPoints[i]).Run() + } + stopCmd(c.filerCmd) + stopCmd(c.volumeCmd) + stopCmd(c.masterCmd) + + for _, f := range c.logFiles { + f.Close() + } + c.copyLogsForCI() + if !c.t.Failed() { + os.RemoveAll(c.baseDir) + } + time.Sleep(2 * time.Second) // let ports drain + }) +} + +// MountDir returns the filesystem path of the i-th mount. +func (c *p2pTestCluster) MountDir(i int) string { return c.mountPoints[i] } + +// masterAddress / filerAddress return SeaweedFS-style addresses encoding +// both ports as "host:httpPort.grpcPort". Without this, downstream +// components fall back to the grpcPort = httpPort + 10000 default, +// which doesn't match the random port we allocate. +func (c *p2pTestCluster) masterAddress() string { + return string(pb.NewServerAddress("127.0.0.1", c.masterPort, c.masterGrpcPort)) +} + +func (c *p2pTestCluster) filerAddress() string { + return string(pb.NewServerAddress("127.0.0.1", c.filerPort, c.filerGrpcPort)) +} + +// MountLog returns the contents of mount i's log file. +func (c *p2pTestCluster) MountLog(i int) string { + return c.tailLogFull(fmt.Sprintf("mount%d", i)) +} + +func (c *p2pTestCluster) startMaster(configDir string) error { + c.masterCmd = exec.Command(c.weedBinary, + "-logdir="+filepath.Join(c.baseDir, "logs"), + "master", + "-ip=127.0.0.1", + "-ip.bind=127.0.0.1", + "-port="+strconv.Itoa(c.masterPort), + "-port.grpc="+strconv.Itoa(c.masterGrpcPort), + "-mdir="+filepath.Join(c.baseDir, "master"), + ) + return c.startCmd(c.masterCmd, "master") +} + +func (c *p2pTestCluster) startVolume(configDir string) error { + volDir := filepath.Join(c.baseDir, "volume") + if err := os.MkdirAll(volDir, 0755); err != nil { + return fmt.Errorf("create volume dir: %w", err) + } + c.volumeCmd = exec.Command(c.weedBinary, + "-logdir="+filepath.Join(c.baseDir, "logs"), + "volume", + "-ip=127.0.0.1", + "-ip.bind=127.0.0.1", + "-port="+strconv.Itoa(c.volumePort), + "-port.grpc="+strconv.Itoa(c.volumeGrpcPort), + "-master="+c.masterAddress(), + "-dir="+volDir, + "-max=10", + ) + return c.startCmd(c.volumeCmd, "volume") +} + +func (c *p2pTestCluster) startFiler(configDir string) error { + filerDir := filepath.Join(c.baseDir, "filer") + if err := os.MkdirAll(filerDir, 0755); err != nil { + return fmt.Errorf("create filer dir: %w", err) + } + c.filerCmd = exec.Command(c.weedBinary, + "-logdir="+filepath.Join(c.baseDir, "logs"), + "filer", + "-ip=127.0.0.1", + "-ip.bind=127.0.0.1", + "-port="+strconv.Itoa(c.filerPort), + "-port.grpc="+strconv.Itoa(c.filerGrpcPort), + "-master="+c.masterAddress(), + "-defaultStoreDir="+filerDir, + // default is -mount.p2p=true; be explicit so the intent is grep-able. + "-mount.p2p=true", + ) + return c.startCmd(c.filerCmd, "filer") +} + +func (c *p2pTestCluster) startMount(idx int, configDir string) error { + cacheDir := filepath.Join(c.baseDir, fmt.Sprintf("cache%d", idx)) + if err := os.MkdirAll(cacheDir, 0755); err != nil { + return fmt.Errorf("create cache dir: %w", err) + } + c.mountCmds[idx] = exec.Command(c.weedBinary, + "-logdir="+filepath.Join(c.baseDir, "logs"), + "-v=4", + "mount", + "-filer="+c.filerAddress(), + "-dir="+c.mountPoints[idx], + "-filer.path=/", + "-dirAutoCreate", + "-allowOthers=false", + "-cacheDir="+cacheDir, + "-peer.enable=true", + fmt.Sprintf("-peer.listen=127.0.0.1:%d", c.mountPeerPorts[idx]), + fmt.Sprintf("-peer.advertise=127.0.0.1:%d", c.mountPeerPorts[idx]), + "-peer.dataCenter=dc1", + fmt.Sprintf("-peer.rack=rack%d", idx), + ) + return c.startCmd(c.mountCmds[idx], fmt.Sprintf("mount%d", idx)) +} + +func (c *p2pTestCluster) startCmd(cmd *exec.Cmd, name string) error { + logPath := filepath.Join(c.baseDir, "logs") + if err := os.MkdirAll(logPath, 0755); err != nil { + return fmt.Errorf("create log dir: %w", err) + } + logFile, err := os.Create(filepath.Join(logPath, name+".log")) + if err != nil { + return err + } + c.logFiles = append(c.logFiles, logFile) + cmd.Stdout = logFile + cmd.Stderr = logFile + return cmd.Start() +} + +func (c *p2pTestCluster) tailLog(name string) string { + data, err := os.ReadFile(filepath.Join(c.baseDir, "logs", name+".log")) + if err != nil { + return fmt.Sprintf("(log %s not available: %v)", name, err) + } + const maxTail = 8192 + if len(data) > maxTail { + data = data[len(data)-maxTail:] + } + return string(data) +} + +func (c *p2pTestCluster) tailLogFull(name string) string { + data, err := os.ReadFile(filepath.Join(c.baseDir, "logs", name+".log")) + if err != nil { + return "" + } + return string(data) +} + +func (c *p2pTestCluster) copyLogsForCI() { + ciLogDir := "/tmp/seaweedfs-fuse-p2p-logs" + os.MkdirAll(ciLogDir, 0755) + logsDir := filepath.Join(c.baseDir, "logs") + entries, err := os.ReadDir(logsDir) + if err != nil { + return + } + for _, e := range entries { + data, err := os.ReadFile(filepath.Join(logsDir, e.Name())) + if err != nil { + continue + } + os.WriteFile(filepath.Join(ciLogDir, e.Name()), data, 0644) + } +} + +// waitForTCP polls addr until it accepts a connection, OR the supplied +// subprocess exits — whichever comes first. Short-circuiting on child +// exit turns a 30 s-spin-on-dead-process into an immediate failure with +// the tail of its log. cmd may be nil for callers that don't track a +// process; in that case we fall back to pure polling. +// +// Liveness is checked with signal 0 (POSIX "is this process alive"). +// We deliberately do NOT call cmd.Wait() in a goroutine here because +// stopCmd() later calls Wait() at shutdown, and only one Wait per +// process is allowed. +func (c *p2pTestCluster) waitForTCP(cmd *exec.Cmd, name, addr string, timeout time.Duration) error { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + conn, err := net.DialTimeout("tcp", addr, time.Second) + if err == nil { + conn.Close() + return nil + } + if cmd != nil && cmd.Process != nil { + // signal 0 doesn't actually send anything; it just asks + // the kernel whether the process exists. An error here + // (typically "process already finished" or ESRCH) means + // the child died before coming up. + if sigErr := cmd.Process.Signal(syscall.Signal(0)); sigErr != nil { + return fmt.Errorf("%s exited before listening on %s: %v\n%s", + name, addr, sigErr, c.tailLog(name)) + } + } + time.Sleep(200 * time.Millisecond) + } + return fmt.Errorf("service at %s not ready within timeout\n%s", addr, c.tailLog(name)) +} + +// waitForMount waits for a FUSE filesystem to actually be mounted by +// watching the device id flip (FUSE mounts have a different Dev than +// their parent dir). +func (c *p2pTestCluster) waitForMount(mountPoint string, timeout time.Duration) error { + parentDir := filepath.Dir(mountPoint) + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + parentStat, err := os.Stat(parentDir) + if err != nil { + time.Sleep(200 * time.Millisecond) + continue + } + mountStat, err := os.Stat(mountPoint) + if err != nil { + time.Sleep(200 * time.Millisecond) + continue + } + parentSys := parentStat.Sys().(*syscall.Stat_t) + mountSys := mountStat.Sys().(*syscall.Stat_t) + if parentSys.Dev != mountSys.Dev { + return nil + } + time.Sleep(200 * time.Millisecond) + } + return fmt.Errorf("mount point %s not ready within timeout (FUSE not detected)", mountPoint) +} + +// --- utilities (binary discovery, process shutdown) --- +// +// Port allocation is delegated to testutil.MustAllocatePorts, which +// holds all listeners open until every port is reserved before +// returning them in a batch — safer than the per-listener +// close-then-reserve pattern fuse_dlm originally used. + +func findWeedBinary() string { + if env := os.Getenv("WEED_BINARY"); env != "" { + if _, err := os.Stat(env); err == nil { + return env + } + } + if p, err := exec.LookPath("weed"); err == nil { + return p + } + return "" +} + +// stopCmd kills a child process with SIGTERM and waits up to 5 s, then +// escalates to SIGKILL. Tolerates nil and already-exited processes. +func stopCmd(cmd *exec.Cmd) { + if cmd == nil || cmd.Process == nil { + return + } + cmd.Process.Signal(syscall.SIGTERM) + done := make(chan struct{}) + go func() { + cmd.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + cmd.Process.Signal(syscall.SIGKILL) + <-done + } +} diff --git a/test/fuse_p2p/peer_chunk_sharing_test.go b/test/fuse_p2p/peer_chunk_sharing_test.go new file mode 100644 index 000000000..c64bc0c7f --- /dev/null +++ b/test/fuse_p2p/peer_chunk_sharing_test.go @@ -0,0 +1,147 @@ +package fuse_p2p + +import ( + "bytes" + "crypto/md5" + "fmt" + "math/rand/v2" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// peerReadSuccessMarker is the log line tryPeerRead emits when a peer +// fetch succeeded. The test greps non-writer mount logs for it to +// prove the p2p path fired. At glog verbosity 4 (framework sets -v=4). +const peerReadSuccessMarker = "peer read successful" + +// seedConvergenceTimeout bounds how long the test waits for: +// - the filer registry to list all mounts, +// - every mount's seed view (via MountList poll) to include all peers, +// - the first announcer flush cycle to publish chunk holders. +// +// With defaults the mount polls MountList every 30 s and flushes +// ChunkAnnounce every 15 s. Allowing 45 s absorbs one MountList refresh +// plus one announce cycle plus some slop for CI variance. +const seedConvergenceTimeout = 90 * time.Second + +// TestPeerChunkSharing_ReadersPullFromPeerCache is the headline p2p +// integration test. It proves at least one non-writer mount can satisfy +// a read from the writer's chunk cache instead of the volume tier. +// +// 1. Bring up 1 master/volume/filer + 3 mounts, all with -peer.enable. +// 2. Mount 0 writes a ~8 MiB file and reads it back so chunks land in +// its local cache and the announcer publishes them. +// 3. Wait for seed convergence + at least one announcer flush cycle. +// 4. BOTH mount 1 and mount 2 read the file. +// +// Why both readers: with 3 mounts, HRW picks one owner for the chunk. +// If that owner is mount 1, only mount 2's read will hit the peer +// path (mount 1's tryPeerRead bails on owner==self). If that owner is +// mount 2, only mount 1's read will. If that owner is mount 0 (the +// writer), both can. So by reading from both, we deterministically +// guarantee at least one non-writer mount exercises the peer path. +// +// Once the peer fetch populates its local cache, subsequent reads +// short-circuit on IsInCache — so we only get one real shot per mount +// per chunk. That's fine: one success is all the test needs. +func TestPeerChunkSharing_ReadersPullFromPeerCache(t *testing.T) { + c := startP2PTestCluster(t, 3) + + // ~8 MiB, pseudo-random so compression doesn't collapse it to one block. + payload := make([]byte, 8*1024*1024) + rng := rand.New(rand.NewPCG(1, 2)) + for i := range payload { + payload[i] = byte(rng.Uint32()) + } + + const relPath = "p2p-test.bin" + writer := c.MountDir(0) + require.NoError(t, os.WriteFile(filepath.Join(writer, relPath), payload, 0644)) + + // Warm mount 0's chunk cache by reading back through its own FUSE. + // Without this the chunks are on the volume server but not yet + // in anyone's peer-servable cache. + readBack, err := os.ReadFile(filepath.Join(writer, relPath)) + require.NoError(t, err) + require.True(t, bytes.Equal(readBack, payload), "write-then-read on writer mount should match") + + waitForSeedConvergence(t, c, seedConvergenceTimeout) + + // Give the announcer several flush windows to push chunk-holder + // entries to the HRW owners. First flush may see the writer's seed + // view incomplete (only self) and defer all fids; subsequent + // flushes re-check against a refreshed seed view. announce interval + // is 15 s, so 45 s covers three attempts. + time.Sleep(45 * time.Second) + + // Read from both non-writer mounts. For any HRW outcome on any + // chunk, at least one of these reads will NOT have the reader as + // the HRW owner, so its tryPeerRead will proceed to ChunkLookup + + // FetchChunk. + for _, idx := range []int{1, 2} { + got, err := os.ReadFile(filepath.Join(c.MountDir(idx), relPath)) + require.NoError(t, err, "read from mount %d must succeed\n--- mount%d ---\n%s", + idx, idx, tailLines(c.MountLog(idx), 80)) + require.Equal(t, md5.Sum(payload), md5.Sum(got), + "mount %d returned mismatched bytes (len got=%d want=%d)", idx, len(got), len(payload)) + } + + // Content matches alone doesn't prove p2p — the volume fallback + // would also satisfy the reads. Require at least one non-writer + // mount's log to contain the peer-read success marker. + var sawPeerRead bool + for _, idx := range []int{1, 2} { + if strings.Contains(c.MountLog(idx), peerReadSuccessMarker) { + sawPeerRead = true + break + } + } + if !sawPeerRead { + t.Fatalf("no non-writer mount logged %q — peer read path never fired.\n"+ + "--- mount0 (writer) tail ---\n%s\n--- mount1 tail ---\n%s\n--- mount2 tail ---\n%s", + peerReadSuccessMarker, + tailLines(c.MountLog(0), 60), tailLines(c.MountLog(1), 60), tailLines(c.MountLog(2), 60)) + } +} + +// waitForSeedConvergence polls each mount's log looking for any sign +// that MountList returned a peer list containing the other mounts. +func waitForSeedConvergence(t *testing.T, c *p2pTestCluster, timeout time.Duration) { + t.Helper() + // The first MountRegister happens synchronously during mount startup, + // and the first MountList is pulled right after. 30 s is the refresh + // interval; waiting one full cycle here guarantees every mount has + // at minimum observed the others in its seed view. + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + ready := true + for i := range c.mountCmds { + if !strings.Contains(c.MountLog(i), "peer-grpc listening on") { + ready = false + break + } + } + if ready { + time.Sleep(30 * time.Second) + return + } + time.Sleep(500 * time.Millisecond) + } + t.Fatalf("peer-grpc servers did not come up within %s", timeout) +} + +// tailLines returns the last n newline-delimited lines of s, or the +// whole thing if shorter. Keeps test failures readable. +func tailLines(s string, n int) string { + lines := strings.Split(s, "\n") + if len(lines) <= n { + return s + } + return fmt.Sprintf("... (%d earlier lines omitted) ...\n%s", + len(lines)-n, strings.Join(lines[len(lines)-n:], "\n")) +} diff --git a/weed/mount/filehandle_read.go b/weed/mount/filehandle_read.go index 48805b60b..df458524d 100644 --- a/weed/mount/filehandle_read.go +++ b/weed/mount/filehandle_read.go @@ -75,6 +75,23 @@ func (fh *FileHandle) readFromChunksWithContext(ctx context.Context, buff []byte glog.V(4).Infof("RDMA read failed for %s, falling back to HTTP: %v", fileFullPath, err) } + // Peer chunk sharing: try a peer mount's cache before the volume tier. + // Any failure falls through transparently. See design-weed-mount- + // peer-chunk-sharing.md §4.3. + if fh.wfs.option.PeerEnabled && fh.wfs.peerGrpcServer != nil { + totalRead, ts, err := fh.tryPeerRead(ctx, fileSize, buff, offset, entry) + if err == nil { + glog.V(4).Infof("peer read successful for %s [%d,%d] %d", fileFullPath, offset, offset+int64(totalRead), totalRead) + return int64(totalRead), ts, nil + } + // Skip the "failed" log for benign skip reasons (local cache + // hit, no peer owner yet, etc.) — the cache/volume fallback is + // the expected outcome, not a failure. + if err != errPeerReadSkipped { + glog.V(4).Infof("peer read failed for %s, falling back to volume: %v", fileFullPath, err) + } + } + // Fall back to normal chunk reading totalRead, ts, err := fh.entryChunkGroup.ReadDataAt(ctx, fileSize, buff, offset) diff --git a/weed/mount/peer_fetcher.go b/weed/mount/peer_fetcher.go new file mode 100644 index 000000000..6354283bc --- /dev/null +++ b/weed/mount/peer_fetcher.go @@ -0,0 +1,324 @@ +package mount + +import ( + "bytes" + "context" + "crypto/md5" + "encoding/hex" + "fmt" + "io" + "sort" + "time" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/mount_peer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// errPeerReadSkipped signals that the read didn't go through the peer +// path for a benign reason (local cache hit, no peer owner, etc.) rather +// than a genuine failure. Callers fall through to the volume path just +// as they would for a real error, but suppress the "peer read failed" +// log that would otherwise mislead operators into thinking peer sharing +// is broken. +var errPeerReadSkipped = fmt.Errorf("peer read skipped (not an error)") + +// peerLookupTimeout bounds the ChunkLookup RPC. Short because the result +// is consumed on the read critical path. +const peerLookupTimeout = 500 * time.Millisecond + +// peerFetchTimeout bounds a single FetchChunk stream; on expiry we fall +// through to the next holder or the volume server. +const peerFetchTimeout = 5 * time.Second + +// maxPeerFetchChunkBytes caps how much we will accept from a single +// FetchChunk stream. gRPC per-message size is already capped by the +// server option; this is belt-and-suspenders against a runaway peer. +const maxPeerFetchChunkBytes = 64 * 1024 * 1024 + +// tryPeerRead attempts to satisfy a read from a peer mount's chunk cache. +// Returns (bytesRead, modifiedTsNs, nil) on success. On any failure it +// returns (0, 0, err) so the caller falls through to +// entryChunkGroup.ReadDataAt (the volume-server path). +// +// Flow: +// 1. Resolve the offset's leaf chunk (flattening manifests). +// 2. Ask the HRW owner mount for current holders via ChunkLookup. +// 3. For each holder (LRU order from PR #5), open a FetchChunk stream, +// assemble frames into a size-bounded buffer, and verify MD5 +// end-to-end against FileChunk.ETag. +// 4. On success, populate chunk_cache and enqueue an announce so +// other mounts can discover us as a new holder. +func (fh *FileHandle) tryPeerRead(ctx context.Context, fileSize int64, buff []byte, offset int64, entry *LockedEntry) (int64, int64, error) { + if fh.wfs.peerRegistrar == nil || fh.wfs.peerConnPool == nil { + return 0, 0, fmt.Errorf("peer sharing not configured") + } + + // Resolve offset → leaf chunk, flattening any manifest indirection. + readStop := offset + int64(len(buff)) + if readStop > fileSize { + readStop = fileSize + } + dataChunks, _, err := filer.ResolveChunkManifest(ctx, fh.wfs.LookupFn(), entry.GetEntry().Chunks, offset, readStop) + if err != nil { + return 0, 0, fmt.Errorf("resolve manifest: %w", err) + } + targetChunk, chunkOffset := findChunkContaining(dataChunks, offset) + if targetChunk == nil { + return 0, 0, fmt.Errorf("no leaf chunk for offset %d", offset) + } + // Reject reads that cross a chunk boundary. We only fetch one chunk + // here, and the caller (readFromChunks) maps a short non-error + // return to "success, zero-fill the rest" — which would silently + // corrupt reads that should actually span two chunks. Bail so the + // fallback ReadDataAt path can handle the multi-chunk case + // correctly. See weedfs_file_read.go short-read semantics. + chunkEnd := targetChunk.Offset + int64(targetChunk.Size) + if readStop > chunkEnd { + return 0, 0, errPeerReadSkipped + } + + // Fail fast when the chunk is already cached locally: the fallback + // ReadDataAt path will satisfy the read from chunkCache with no RPCs, + // so dialing a peer (ChunkLookup + FetchChunk) would be pure overhead. + if fh.wfs.chunkCache != nil && fh.wfs.chunkCache.IsInCache(targetChunk.FileId, true) { + return 0, 0, errPeerReadSkipped + } + + selfAddr := "" + if fh.wfs.peerGrpcServer != nil { + selfAddr = fh.wfs.peerGrpcServer.SelfAddr() + } + + owner := fh.wfs.peerRegistrar.OwnerFor(targetChunk.FileId) + if owner == "" || owner == selfAddr { + return 0, 0, fmt.Errorf("no peer owner for fid %s", targetChunk.FileId) + } + + holders, err := peerLookupHolders(ctx, fh.wfs.peerConnPool.Dialer(), owner, targetChunk.FileId) + if err != nil { + return 0, 0, fmt.Errorf("peer lookup: %w", err) + } + if len(holders) == 0 { + return 0, 0, fmt.Errorf("no peer holder for fid %s", targetChunk.FileId) + } + + // Re-rank holders by locality (same rack > same DC > elsewhere), keeping + // the server's LRU order stable within each bucket. The server caps the + // list at maxLookupHolders, so this is always a small N. + sortHoldersByLocality(holders, fh.wfs.option.PeerDataCenter, fh.wfs.option.PeerRack) + + dial := fh.wfs.peerConnPool.Dialer() + for _, h := range holders { + if h.addr == selfAddr { + continue + } + data, ferr := fetchChunkFromPeer(ctx, dial, h.addr, targetChunk.FileId, targetChunk.Size, targetChunk.ETag) + if ferr != nil { + glog.V(2).Infof("peer-fetch %s from %s: %v", targetChunk.FileId, h.addr, ferr) + continue + } + if fh.wfs.chunkCache != nil { + fh.wfs.chunkCache.SetChunk(targetChunk.FileId, data) + } + if fh.wfs.peerAnnouncer != nil { + fh.wfs.peerAnnouncer.EnqueueAnnounce(targetChunk.FileId) + } + if chunkOffset >= int64(len(data)) { + return 0, 0, fmt.Errorf("peer returned short chunk") + } + // Cap the copy to whichever is smaller: the caller's buffer, or + // the remaining bytes before logical EOF (readStop already + // clamped to fileSize). FileChunk.Size can legitimately exceed + // the logical file length when the last chunk is partially + // written and the filer stored the full padded buffer, so a + // naïve copy(buff, data[chunkOffset:]) would return bytes past + // EOF. The caller treats a short non-error return as success + // and zero-fills the tail, so returning too many bytes here is + // a correctness bug, not just wasted I/O. + remaining := readStop - offset + available := int64(len(data)) - chunkOffset + maxCopy := int64(len(buff)) + if remaining < maxCopy { + maxCopy = remaining + } + if available < maxCopy { + maxCopy = available + } + copied := copy(buff[:maxCopy], data[chunkOffset:chunkOffset+maxCopy]) + return int64(copied), targetChunk.ModifiedTsNs, nil + } + return 0, 0, fmt.Errorf("no peer served fid %s", targetChunk.FileId) +} + +func findChunkContaining(chunks []*filer_pb.FileChunk, offset int64) (*filer_pb.FileChunk, int64) { + for _, c := range chunks { + start := c.Offset + stop := c.Offset + int64(c.Size) + if offset >= start && offset < stop { + return c, offset - start + } + } + return nil, 0 +} + +// peerHolder is a holder entry carried through the fetcher: addr for +// dialing plus the DC/Rack labels the owner recorded at announce time, so +// the fetcher can re-rank by its own locality before dialing. +type peerHolder struct { + addr string + dc string + rack string +} + +// peerLookupHolders calls ChunkLookup on the given HRW owner and returns +// the holders in the server-reported (LRU) order along with their locality +// labels. The pooled dialer is passed in from the caller. +func peerLookupHolders(ctx context.Context, dial MountPeerDialer, ownerAddr, fid string) ([]peerHolder, error) { + client, closeFn, err := dial(ctx, ownerAddr) + if err != nil { + return nil, err + } + defer closeFn() + + callCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout) + defer cancel() + resp, err := client.ChunkLookup(callCtx, &mount_peer_pb.ChunkLookupRequest{FileIds: []string{fid}}) + if err != nil { + return nil, err + } + set, ok := resp.PeersByFid[fid] + if !ok || set == nil { + return nil, nil + } + out := make([]peerHolder, 0, len(set.Peers)) + for _, p := range set.Peers { + out = append(out, peerHolder{addr: p.PeerAddr, dc: p.DataCenter, rack: p.Rack}) + } + return out, nil +} + +// localityBucket scores how "close" a peer is to self. Lower is better: +// +// 0 = same rack in same DC (shortest hop) +// 1 = same DC, different rack (cross-rack but still intra-DC) +// 2 = different DC, or unknown (anything else) +// +// Missing labels on either side fall to bucket 2 — we don't claim locality +// we can't prove. +func localityBucket(selfDC, selfRack, peerDC, peerRack string) int { + if selfDC == "" || peerDC == "" || selfDC != peerDC { + return 2 + } + if selfRack != "" && peerRack != "" && selfRack == peerRack { + return 0 + } + return 1 +} + +// sortHoldersByLocality stable-sorts holders so the most local peers are +// tried first. The server-returned order is LRU (freshest holder first), +// and sort.SliceStable preserves that ordering within each locality +// bucket, so among equally-local peers we still prefer the freshest. +func sortHoldersByLocality(holders []peerHolder, selfDC, selfRack string) { + if len(holders) < 2 { + return + } + sort.SliceStable(holders, func(i, j int) bool { + return localityBucket(selfDC, selfRack, holders[i].dc, holders[i].rack) < + localityBucket(selfDC, selfRack, holders[j].dc, holders[j].rack) + }) +} + +// fetchChunkFromPeer server-streams a chunk from the given peer and +// verifies the MD5 of the assembled bytes against expectedETag. +// +// expectedSize (when > 0) is the authoritative chunk length from the +// filer entry; we pre-allocate exactly that to avoid slice growth and +// reject streams that overshoot it. A zero expectedSize falls back to +// maxPeerFetchChunkBytes as a safety ceiling. +func fetchChunkFromPeer(ctx context.Context, dial MountPeerDialer, peerAddr, fid string, expectedSize uint64, expectedETag string) ([]byte, error) { + client, closeFn, err := dial(ctx, peerAddr) + if err != nil { + return nil, err + } + defer closeFn() + + callCtx, cancel := context.WithTimeout(ctx, peerFetchTimeout) + defer cancel() + + stream, err := client.FetchChunk(callCtx, &mount_peer_pb.FetchChunkRequest{ + FileId: fid, + ExpectedEtag: expectedETag, + ExpectedSize: expectedSize, + }) + if err != nil { + return nil, err + } + + // capHint pre-sizes the assembly buffer from the filer-reported chunk + // size. When the caller didn't know the size (expectedSize == 0), we + // let append grow the buffer rather than reserve the 64 MiB ceiling — + // typical chunks are a few MiB, and the maxPeerFetchChunkBytes check + // during Recv is the real safety ceiling. + capHint := expectedSize + if capHint > maxPeerFetchChunkBytes { + capHint = maxPeerFetchChunkBytes + } + buf := make([]byte, 0, capHint) + + for { + resp, rerr := stream.Recv() + if rerr == io.EOF { + break + } + if rerr != nil { + if status.Code(rerr) == codes.NotFound { + return nil, fmt.Errorf("peer not cached") + } + return nil, rerr + } + if len(buf)+len(resp.Data) > int(maxPeerFetchChunkBytes) { + return nil, fmt.Errorf("peer response exceeds max chunk size %d", maxPeerFetchChunkBytes) + } + buf = append(buf, resp.Data...) + } + + if expectedSize > 0 && uint64(len(buf)) != expectedSize { + return nil, fmt.Errorf("peer returned %d bytes, expected %d", len(buf), expectedSize) + } + + // Per-chunk integrity check — the peer is treated as untrusted. + // FileChunk.ETag is UploadResult.ContentMd5, which is base64 of the + // raw 16-byte MD5 (see weed/operation/upload_content.go:64 and + // filer's ETagChunks which decodes via util.Base64Md5ToBytes). + // Compare raw digests so we don't reject every valid peer response + // because of an encoding mismatch. + if expectedETag != "" { + got := md5.Sum(buf) + if !etagMatchesMD5(expectedETag, got[:]) { + return nil, fmt.Errorf("etag mismatch: peer=%s got=%x want=%s", peerAddr, got, expectedETag) + } + } + return buf, nil +} + +// etagMatchesMD5 compares a stored FileChunk.ETag string to a raw 16-byte +// MD5 digest. FileChunk.ETag is produced as base64(md5) on upload, but +// some older chunks (and tests) use hex(md5). Accept either. +func etagMatchesMD5(etag string, rawMD5 []byte) bool { + if len(rawMD5) != md5.Size { + return false + } + if dec := util.Base64Md5ToBytes(etag); len(dec) == md5.Size { + return bytes.Equal(dec, rawMD5) + } + if dec, err := hex.DecodeString(etag); err == nil && len(dec) == md5.Size { + return bytes.Equal(dec, rawMD5) + } + return false +} diff --git a/weed/mount/peer_fetcher_test.go b/weed/mount/peer_fetcher_test.go new file mode 100644 index 000000000..867093941 --- /dev/null +++ b/weed/mount/peer_fetcher_test.go @@ -0,0 +1,180 @@ +package mount + +import ( + "context" + "crypto/md5" + "encoding/base64" + "encoding/hex" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// newTestFetchServer stands up a minimal MountPeer gRPC server that only +// serves FetchChunk out of a fakeChunkCache. Goes through the same +// PeerGrpcServer.Start path production uses (via pb.NewGrpcServer with +// the standard keepalive + msg-size options) so future default server +// options are exercised by these tests too. +func newTestFetchServer(t *testing.T, cache *fakeChunkCache) (addr string, stop func()) { + t.Helper() + dir := NewPeerDirectory() + srv := NewPeerGrpcServer(cache, dir, nil, "") + if err := srv.Start("127.0.0.1:0"); err != nil { + t.Fatalf("start: %v", err) + } + return srv.Addr(), func() { srv.Stop() } +} + +func etagOf(b []byte) string { + sum := md5.Sum(b) + return hex.EncodeToString(sum[:]) +} + +// etagOfBase64 mirrors how the real upload path produces FileChunk.ETag: +// base64 of the raw 16-byte MD5 digest. Kept alongside the hex form so +// we can test that both encodings work end-to-end. +func etagOfBase64(b []byte) string { + sum := md5.Sum(b) + return base64.StdEncoding.EncodeToString(sum[:]) +} + +func TestFetchChunkFromPeer_Hit(t *testing.T) { + cache := newFakeChunkCache() + payload := []byte("hello from peer stream") + cache.Put("3,abc", payload) + addr, stop := newTestFetchServer(t, cache) + defer stop() + + dial := DefaultMountPeerDialer(grpc.WithTransportCredentials(insecure.NewCredentials())) + got, err := fetchChunkFromPeer(context.Background(), dial, addr, "3,abc", uint64(len(payload)), etagOf(payload)) + if err != nil { + t.Fatalf("fetch: %v", err) + } + if string(got) != string(payload) { + t.Errorf("bytes mismatch: got %q want %q", got, payload) + } +} + +// TestFetchChunkFromPeer_Base64Etag mirrors the real upload path where +// FileChunk.ETag is base64(md5) rather than hex(md5). A regression +// there would reject every valid peer response in production. +func TestFetchChunkFromPeer_Base64Etag(t *testing.T) { + cache := newFakeChunkCache() + payload := []byte("base64-etag peer bytes") + cache.Put("3,b64", payload) + addr, stop := newTestFetchServer(t, cache) + defer stop() + + dial := DefaultMountPeerDialer(grpc.WithTransportCredentials(insecure.NewCredentials())) + got, err := fetchChunkFromPeer(context.Background(), dial, addr, "3,b64", uint64(len(payload)), etagOfBase64(payload)) + if err != nil { + t.Fatalf("fetch: %v", err) + } + if string(got) != string(payload) { + t.Errorf("bytes mismatch: got %q want %q", got, payload) + } +} + +func TestFetchChunkFromPeer_EtagMismatch(t *testing.T) { + cache := newFakeChunkCache() + payload := []byte("unexpected bytes") + cache.Put("3,abc", payload) + addr, stop := newTestFetchServer(t, cache) + defer stop() + + dial := DefaultMountPeerDialer(grpc.WithTransportCredentials(insecure.NewCredentials())) + _, err := fetchChunkFromPeer(context.Background(), dial, addr, "3,abc", 0, "not-the-real-etag") + if err == nil { + t.Fatalf("expected etag mismatch error, got nil") + } +} + +func TestFetchChunkFromPeer_NotFound(t *testing.T) { + cache := newFakeChunkCache() + addr, stop := newTestFetchServer(t, cache) + defer stop() + + dial := DefaultMountPeerDialer(grpc.WithTransportCredentials(insecure.NewCredentials())) + _, err := fetchChunkFromPeer(context.Background(), dial, addr, "3,missing", 0, "") + if err == nil { + t.Errorf("expected error for missing fid, got nil") + } +} + +// TestSortHoldersByLocality verifies that same-rack peers sort ahead of +// same-DC-different-rack peers, which in turn sort ahead of cross-DC +// peers, and that LRU order is preserved within each bucket. +func TestSortHoldersByLocality(t *testing.T) { + selfDC, selfRack := "dc1", "r1" + // Input order mimics a server-side LRU list (newest first). + holders := []peerHolder{ + {addr: "far-newest", dc: "dc2", rack: "r9"}, // bucket 2 (diff DC) + {addr: "mid-newer", dc: "dc1", rack: "r2"}, // bucket 1 (same DC, diff rack) + {addr: "local-newer", dc: "dc1", rack: "r1"}, // bucket 0 (same rack) + {addr: "far-older", dc: "dc2", rack: "r9"}, // bucket 2 + {addr: "mid-older", dc: "dc1", rack: "r2"}, // bucket 1 + {addr: "local-older", dc: "dc1", rack: "r1"}, // bucket 0 + {addr: "unlabeled-older", dc: "", rack: ""}, // bucket 2 (unknown) + } + + sortHoldersByLocality(holders, selfDC, selfRack) + + want := []string{ + "local-newer", "local-older", + "mid-newer", "mid-older", + "far-newest", "far-older", "unlabeled-older", + } + if len(holders) != len(want) { + t.Fatalf("len got %d want %d", len(holders), len(want)) + } + for i, w := range want { + if holders[i].addr != w { + t.Errorf("pos %d: got %q want %q (full: %+v)", i, holders[i].addr, w, holders) + } + } +} + +// TestSortHoldersByLocality_NoSelfLabels — when the caller has no DC/rack +// labels, every peer falls to bucket 2 and the server-returned LRU order +// must pass through unchanged. +func TestSortHoldersByLocality_NoSelfLabels(t *testing.T) { + holders := []peerHolder{ + {addr: "a", dc: "dc1", rack: "r1"}, + {addr: "b", dc: "dc2", rack: "r2"}, + {addr: "c", dc: "", rack: ""}, + } + sortHoldersByLocality(holders, "", "") + want := []string{"a", "b", "c"} + for i, w := range want { + if holders[i].addr != w { + t.Errorf("pos %d: got %q want %q", i, holders[i].addr, w) + } + } +} + +func TestFetchChunkFromPeer_MultiFrameChunkAssembledCorrectly(t *testing.T) { + cache := newFakeChunkCache() + // Just over 2× the stream frame size so we get at least three frames. + payload := make([]byte, fetchChunkStreamSize*2+42) + for i := range payload { + payload[i] = byte((i * 7) & 0xff) + } + cache.Put("3,large", payload) + addr, stop := newTestFetchServer(t, cache) + defer stop() + + dial := DefaultMountPeerDialer(grpc.WithTransportCredentials(insecure.NewCredentials())) + got, err := fetchChunkFromPeer(context.Background(), dial, addr, "3,large", uint64(len(payload)), etagOf(payload)) + if err != nil { + t.Fatalf("fetch: %v", err) + } + if len(got) != len(payload) { + t.Fatalf("len got %d want %d", len(got), len(payload)) + } + for i := range payload { + if got[i] != payload[i] { + t.Fatalf("mismatch at offset %d", i) + } + } +} diff --git a/weed/mount/peer_grpc.go b/weed/mount/peer_grpc.go index 7f8280568..9cabe48f7 100644 --- a/weed/mount/peer_grpc.go +++ b/weed/mount/peer_grpc.go @@ -108,6 +108,10 @@ func (s *PeerGrpcServer) Addr() string { return s.listener.Addr().String() } +// SelfAddr returns the advertise address this server was constructed with — +// the identity the fetcher compares against to avoid dialing itself. +func (s *PeerGrpcServer) SelfAddr() string { return s.selfAddr } + // ChunkAnnounce accepts holder entries for fids this mount owns; rejects // others so the caller can retry against the correct owner. func (s *PeerGrpcServer) ChunkAnnounce(ctx context.Context, req *mount_peer_pb.ChunkAnnounceRequest) (*mount_peer_pb.ChunkAnnounceResponse, error) {