diff --git a/.github/workflows/fuse-failover.yml b/.github/workflows/fuse-failover.yml new file mode 100644 index 000000000..de79f6c68 --- /dev/null +++ b/.github/workflows/fuse-failover.yml @@ -0,0 +1,75 @@ +name: "FUSE Volume Server Failover Tests" + +on: + pull_request: + paths: + - 'weed/command/mount*.go' + - 'weed/mount/**' + - 'weed/filer/**' + - 'weed/wdclient/**' + - 'weed/operation/upload_content.go' + - 'test/fuse_failover/**' + - '.github/workflows/fuse-failover.yml' + - '.github/actions/fix-fusermount-setuid/**' + push: + branches: [master] + paths: + - 'weed/command/mount*.go' + - 'weed/mount/**' + - 'weed/filer/**' + - 'weed/wdclient/**' + - 'weed/operation/upload_content.go' + - 'test/fuse_failover/**' + - '.github/workflows/fuse-failover.yml' + - '.github/actions/fix-fusermount-setuid/**' + +concurrency: + group: ${{ github.head_ref || github.ref }}/fuse-failover + cancel-in-progress: true + +permissions: + contents: read + +jobs: + fuse-failover: + name: FUSE Volume Server Failover + runs-on: ubuntu-22.04 + timeout-minutes: 40 + + steps: + - name: Check out code + uses: actions/checkout@v7 + with: + persist-credentials: false + + - name: Set up Go + uses: actions/setup-go@v7 + 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: Repair the fusermount3 setuid bit + uses: ./.github/actions/fix-fusermount-setuid + + - name: Build SeaweedFS + run: go build -o weed/weed -buildvcs=false ./weed + + - name: Run failover integration tests + timeout-minutes: 35 + env: + WEED_BINARY: ${{ github.workspace }}/weed/weed + run: go test -v -count=1 -timeout=30m ./test/fuse_failover/... + + - name: Upload logs on failure + if: failure() + uses: actions/upload-artifact@v7 + with: + name: fuse-failover-test-logs + path: /tmp/seaweedfs-fuse-failover-logs/ + retention-days: 3 diff --git a/test/fuse_failover/README.md b/test/fuse_failover/README.md new file mode 100644 index 000000000..422145b8b --- /dev/null +++ b/test/fuse_failover/README.md @@ -0,0 +1,35 @@ +# FUSE volume server failover tests + +Integration tests for what happens to FUSE mounts when a volume server goes +away, comes back, or restarts underneath in-flight IO. They automate the manual +matrix reported in +[discussion #10206](https://github.com/seaweedfs/seaweedfs/discussions/10206): +one mount appends to a file while a second mount tails it, and a volume server +is stopped, started or restarted mid-stream. + +The cluster is 1 master (`-defaultReplication=001`), 3 volume servers, 1 filer +and 2 mounts, all as local processes. With 001 every chunk has a copy on two of +the three servers, so losing any single server must be invisible to both mounts. + +| Reported scenario | Test | +| --- | --- | +| control: append + tail with nothing failing | `TestAppendWithoutChaos` | +| read a file while one volume server is down | `TestReadWithVolumeServerDown` | +| "STOP volumes": append + tail, kill a server mid-stream | `TestAppendWhileVolumeServerStops` | +| "Start volumes": append + tail with a server down, start it mid-stream | `TestAppendWhileVolumeServerStarts` | +| "Re-start volumes": append + tail, restart a server mid-stream | `TestAppendWhileVolumeServerRestarts` | +| part 2: large file copy instead of small appends | `TestLargeWriteWhileVolumeServerStops` | + +Volume servers are dropped with SIGKILL, the closest local equivalent of a Swarm +task disappearing from the overlay network: no deregistration, and the address +stops answering. + +## Running + +```bash +go build -o weed/weed ./weed +WEED_BINARY=$PWD/weed/weed go test -v -count=1 -timeout=30m ./test/fuse_failover/... +``` + +Needs FUSE and, on Linux, a working `/dev/fuse`. Logs from a failed run are +copied to `/tmp/seaweedfs-fuse-failover-logs/`. diff --git a/test/fuse_failover/framework_test.go b/test/fuse_failover/framework_test.go new file mode 100644 index 000000000..28c26a250 --- /dev/null +++ b/test/fuse_failover/framework_test.go @@ -0,0 +1,537 @@ +//go:build linux || darwin + +package fuse_failover + +import ( + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/testutil" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/stretchr/testify/require" +) + +// failoverCluster runs 1 master, N volume servers, 1 filer and M FUSE mounts, +// with the master defaulting to 001 replication so every chunk lands on two +// distinct volume servers. Individual volume servers can be stopped and started +// while IO is in flight, which is what the Docker Swarm reports in discussion +// 10206 exercise: a volume server disappears mid-append and the mounts must +// keep reading from the surviving replica and keep writing on another volume. +type failoverCluster struct { + t testing.TB + baseDir string + weedBinary string + + masterPort int + masterGrpcPort int + filerPort int + filerGrpcPort int + volumePorts []int + volumeGrpcPort []int + volumeDirs []string + + masterCmd *exec.Cmd + filerCmd *exec.Cmd + volumeCmds []*exec.Cmd + mountCmds []*exec.Cmd + mountPoints []string + logFiles []*os.File + + mu sync.Mutex + waits map[*exec.Cmd]chan error + cleanupOnce sync.Once +} + +func startFailoverCluster(t testing.TB, numVolumes, numMounts int) *failoverCluster { + require.GreaterOrEqual(t, numVolumes, 2, "001 replication needs at least 2 volume servers") + require.GreaterOrEqual(t, numMounts, 1) + + 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_failover_") + require.NoError(t, err) + + c := &failoverCluster{ + t: t, + baseDir: baseDir, + weedBinary: binary, + volumePorts: make([]int, numVolumes), + volumeGrpcPort: make([]int, numVolumes), + volumeDirs: make([]string, numVolumes), + volumeCmds: make([]*exec.Cmd, numVolumes), + mountCmds: make([]*exec.Cmd, numMounts), + mountPoints: make([]string, numMounts), + } + t.Cleanup(c.Stop) + + ports, err := testutil.AllocatePorts(4 + 2*numVolumes) + require.NoError(t, err) + c.masterPort, c.masterGrpcPort = ports[0], ports[1] + c.filerPort, c.filerGrpcPort = ports[2], ports[3] + for i := 0; i < numVolumes; i++ { + c.volumePorts[i] = ports[4+2*i] + c.volumeGrpcPort[i] = ports[5+2*i] + c.volumeDirs[i] = filepath.Join(baseDir, fmt.Sprintf("volume%d", i)) + require.NoError(t, os.MkdirAll(c.volumeDirs[i], 0755)) + } + + require.NoError(t, c.startMaster()) + require.NoError(t, c.waitForTCP(c.masterCmd, "master", + fmt.Sprintf("127.0.0.1:%d", c.masterPort), 30*time.Second)) + + for i := 0; i < numVolumes; i++ { + require.NoError(t, c.StartVolume(i)) + } + + require.NoError(t, c.startFiler()) + 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)) + 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 *failoverCluster) MountDir(i int) string { return c.mountPoints[i] } + +func (c *failoverCluster) Stop() { + if c == nil { + return + } + c.cleanupOnce.Do(func() { + for i := len(c.mountCmds) - 1; i >= 0; i-- { + c.stopCmd(c.mountCmds[i], syscall.SIGTERM) + _ = exec.Command("fusermount3", "-u", c.mountPoints[i]).Run() + _ = exec.Command("fusermount", "-u", c.mountPoints[i]).Run() + } + c.stopCmd(c.filerCmd, syscall.SIGTERM) + for i := len(c.volumeCmds) - 1; i >= 0; i-- { + c.stopCmd(c.volumeCmds[i], syscall.SIGTERM) + } + c.stopCmd(c.masterCmd, syscall.SIGTERM) + + c.mu.Lock() + for _, f := range c.logFiles { + _ = f.Close() + } + c.mu.Unlock() + c.copyLogsForCI() + if !c.t.Failed() { + os.RemoveAll(c.baseDir) + } + }) +} + +// KillVolume drops a volume server without letting it deregister, the closest +// local equivalent of a Swarm task vanishing from the overlay network. +func (c *failoverCluster) KillVolume(i int) { + c.stopCmd(c.volumeCmds[i], syscall.SIGKILL) + c.volumeCmds[i] = nil +} + +// StartVolume (re)starts volume server i on its original ports and data dir. +func (c *failoverCluster) StartVolume(i int) error { + cmd := 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.volumePorts[i]), + "-port.grpc="+strconv.Itoa(c.volumeGrpcPort[i]), + "-master="+c.masterAddress(), + "-dir="+c.volumeDirs[i], + "-dataCenter=dc1", + "-rack=rack1", + "-max=10", + ) + c.volumeCmds[i] = cmd + if err := c.startCmd(cmd, fmt.Sprintf("volume%d", i)); err != nil { + return err + } + return c.waitForTCP(cmd, fmt.Sprintf("volume%d", i), + fmt.Sprintf("127.0.0.1:%d", c.volumePorts[i]), 30*time.Second) +} + +func (c *failoverCluster) startMaster() 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"), + "-defaultReplication=001", + "-volumeSizeLimitMB=64", + ) + return c.startCmd(c.masterCmd, "master") +} + +func (c *failoverCluster) startFiler() 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(), + "-defaultReplicaPlacement=001", + "-defaultStoreDir="+filerDir, + ) + return c.startCmd(c.filerCmd, "filer") +} + +func (c *failoverCluster) startMount(idx int) 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) + } + // Chunk-level detail needs -v=4; keep CI at -v=2 so the logs stay small. + verbosity := os.Getenv("FUSE_FAILOVER_MOUNT_V") + if verbosity == "" { + verbosity = "2" + } + c.mountCmds[idx] = exec.Command(c.weedBinary, + "-logdir="+filepath.Join(c.baseDir, "logs"), + "-v="+verbosity, + "mount", + "-filer="+c.filerAddress(), + "-dir="+c.mountPoints[idx], + "-filer.path=/", + "-dirAutoCreate", + "-allowOthers=false", + "-replication=001", + "-cacheDir="+cacheDir, + ) + return c.startCmd(c.mountCmds[idx], fmt.Sprintf("mount%d", idx)) +} + +// MasterGet fetches a master HTTP endpoint, e.g. "/dir/status?pretty=y" or +// "/dir/lookup?volumeId=6", so a failing test can show where the replicas are. +func (c *failoverCluster) MasterGet(path string) string { + resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d%s", c.masterPort, path)) + if err != nil { + return fmt.Sprintf("(master %s failed: %v)", path, err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Sprintf("(master %s read failed: %v)", path, err) + } + return string(body) +} + +// FilerGet reads a file back through the filer's own HTTP handler: a view of +// the chunk list that neither mount's cache can colour. +func (c *failoverCluster) FilerGet(path string) ([]byte, error) { + resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d%s", c.filerPort, path)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("filer %s: %s", path, resp.Status) + } + return body, nil +} + +// VolumeServerAddress is the address volume server i registers with the master. +func (c *failoverCluster) VolumeServerAddress(i int) string { + return fmt.Sprintf("127.0.0.1:%d", c.volumePorts[i]) +} + +// FileVolumeIds returns the volume ids backing a file, read from the filer's +// own entry rather than inferred, so a test can tell which servers a given +// file actually depends on. Manifests are resolved first: a manifest chunk's +// own fid names the volume holding the manifest, not the data. +func (c *failoverCluster) FileVolumeIds(path string) ([]uint32, error) { + body, err := c.FilerGet(path + "?metadata=true&resolveManifest=true") + if err != nil { + return nil, err + } + var entry struct { + Chunks []struct { + FileId string `json:"file_id"` + Fid struct { + VolumeId uint32 `json:"volume_id"` + } `json:"fid"` + } `json:"chunks"` + } + if err = json.Unmarshal(body, &entry); err != nil { + return nil, fmt.Errorf("decode entry %s: %w", path, err) + } + seen := make(map[uint32]bool) + var vids []uint32 + for _, chunk := range entry.Chunks { + vid := chunk.Fid.VolumeId + if vid == 0 && chunk.FileId != "" { + parsed, parseErr := strconv.ParseUint(strings.SplitN(chunk.FileId, ",", 2)[0], 10, 32) + if parseErr != nil { + return nil, fmt.Errorf("parse file id %s: %w", chunk.FileId, parseErr) + } + vid = uint32(parsed) + } + if !seen[vid] { + seen[vid] = true + vids = append(vids, vid) + } + } + return vids, nil +} + +// VolumeHolders returns the volume server addresses the master currently lists +// for a volume id. +func (c *failoverCluster) VolumeHolders(vid uint32) ([]string, error) { + var lookup struct { + Locations []struct { + Url string `json:"url"` + } `json:"locations"` + } + body := c.MasterGet(fmt.Sprintf("/dir/lookup?volumeId=%d", vid)) + if err := json.Unmarshal([]byte(body), &lookup); err != nil { + return nil, fmt.Errorf("decode lookup for volume %d: %w (%s)", vid, err, body) + } + holders := make([]string, 0, len(lookup.Locations)) + for _, loc := range lookup.Locations { + holders = append(holders, loc.Url) + } + return holders, nil +} + +// WaitForHolders polls the master until it lists exactly count servers for a +// volume. The master only drops a dead node after three missed heartbeats, so a +// test that depends on the cluster's view having caught up has to wait for it. +func (c *failoverCluster) WaitForHolders(vid uint32, count int, timeout time.Duration) ([]string, error) { + deadline := time.Now().Add(timeout) + for { + holders, err := c.VolumeHolders(vid) + if err == nil && len(holders) == count { + return holders, nil + } + if time.Now().After(deadline) { + return holders, fmt.Errorf("volume %d still has %d holders (%v), want %d", vid, len(holders), holders, count) + } + time.Sleep(500 * time.Millisecond) + } +} + +// volumeIndexOf maps a server address back to its index, or -1. +func (c *failoverCluster) volumeIndexOf(address string) int { + for i := range c.volumePorts { + if c.VolumeServerAddress(i) == address { + return i + } + } + return -1 +} + +// FileIsOn reports whether any of path's chunks live on the given volume +// server, i.e. whether taking that server down actually costs this file a +// replica. +func (c *failoverCluster) FileIsOn(path, serverAddress string) (bool, error) { + vids, err := c.FileVolumeIds(path) + if err != nil { + return false, err + } + for _, vid := range vids { + holders, holdersErr := c.VolumeHolders(vid) + if holdersErr != nil { + return false, holdersErr + } + for _, holder := range holders { + if holder == serverAddress { + return true, nil + } + } + } + return false, nil +} + +func (c *failoverCluster) masterAddress() string { + return string(pb.NewServerAddress("127.0.0.1", c.masterPort, c.masterGrpcPort)) +} + +func (c *failoverCluster) filerAddress() string { + return string(pb.NewServerAddress("127.0.0.1", c.filerPort, c.filerGrpcPort)) +} + +func (c *failoverCluster) 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) + } + // Append so a restarted volume server keeps the log of its earlier run. + logFile, err := os.OpenFile(filepath.Join(logPath, name+".log"), + os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + if err != nil { + return err + } + c.mu.Lock() + c.logFiles = append(c.logFiles, logFile) + c.mu.Unlock() + cmd.Stdout = logFile + cmd.Stderr = logFile + if err := cmd.Start(); err != nil { + return err + } + // Reap in the background and publish the result: Signal(0) succeeds for a + // zombie, so an unreaped child that died at startup would otherwise look + // alive until the readiness timeout expired. + ch := make(chan error, 1) + c.mu.Lock() + if c.waits == nil { + c.waits = make(map[*exec.Cmd]chan error) + } + c.waits[cmd] = ch + c.mu.Unlock() + go func() { + ch <- cmd.Wait() + close(ch) + }() + return nil +} + +// waitChan returns the channel carrying cmd's exit, or nil if it was never +// started through startCmd. It stays readable after the exit is consumed. +func (c *failoverCluster) waitChan(cmd *exec.Cmd) chan error { + if cmd == nil { + return nil + } + c.mu.Lock() + defer c.mu.Unlock() + return c.waits[cmd] +} + +func (c *failoverCluster) 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 *failoverCluster) copyLogsForCI() { + // One directory per test: subtests share a log dir name otherwise, and the + // last one to finish would overwrite the logs of the one that failed. + ciLogDir := filepath.Join("/tmp/seaweedfs-fuse-failover-logs", + strings.ReplaceAll(c.t.Name(), "/", "_")) + os.MkdirAll(ciLogDir, 0755) + entries, err := os.ReadDir(filepath.Join(c.baseDir, "logs")) + if err != nil { + return + } + for _, e := range entries { + data, err := os.ReadFile(filepath.Join(c.baseDir, "logs", e.Name())) + if err != nil { + continue + } + os.WriteFile(filepath.Join(ciLogDir, e.Name()), data, 0644) + } +} + +func (c *failoverCluster) 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 ch := c.waitChan(cmd); ch != nil { + select { + case waitErr := <-ch: + return fmt.Errorf("%s exited before listening on %s: %v\n%s", + name, addr, waitErr, c.tailLog(name)) + default: + } + } + time.Sleep(200 * time.Millisecond) + } + return fmt.Errorf("service at %s not ready within timeout\n%s", addr, c.tailLog(name)) +} + +func (c *failoverCluster) 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 + } + if parentStat.Sys().(*syscall.Stat_t).Dev != mountStat.Sys().(*syscall.Stat_t).Dev { + return nil + } + time.Sleep(200 * time.Millisecond) + } + return fmt.Errorf("mount point %s not ready within timeout (FUSE not detected)", mountPoint) +} + +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 signals cmd and waits for the reaper goroutine started by startCmd to +// report its exit, escalating to SIGKILL if it does not go quietly. +func (c *failoverCluster) stopCmd(cmd *exec.Cmd, sig syscall.Signal) { + if cmd == nil || cmd.Process == nil { + return + } + _ = cmd.Process.Signal(sig) + done := c.waitChan(cmd) + if done == nil { + return + } + select { + case <-done: + case <-time.After(10 * time.Second): + _ = cmd.Process.Signal(syscall.SIGKILL) + <-done + } +} diff --git a/test/fuse_failover/volume_failover_test.go b/test/fuse_failover/volume_failover_test.go new file mode 100644 index 000000000..60dc5b7cf --- /dev/null +++ b/test/fuse_failover/volume_failover_test.go @@ -0,0 +1,419 @@ +//go:build linux || darwin + +package fuse_failover + +import ( + "bytes" + "crypto/rand" + "crypto/sha256" + "fmt" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// The scenarios below come from the Docker Swarm report in discussion 10206: +// one mount appends to a file while a second mount reads it, and a volume +// server is stopped, started or restarted underneath. With 001 replication a +// single volume server loss must never surface as EIO on either side. + +const ( + appendLines = 200 + maxAppendLatency = 10 * time.Second + convergeTimeout = 30 * time.Second +) + +type appendResult struct { + errs []error + maxLatency time.Duration + slowest int +} + +// appendLoop mimics `for i in ...; do echo $i >> file; done`: every line is its +// own open/write/close, so every line forces a flush and a chunk upload. +func appendLoop(path string, lines int, onLine func(i int)) *appendResult { + res := &appendResult{} + for i := 0; i < lines; i++ { + start := time.Now() + err := appendOnce(path, fmt.Sprintf("%07d\n", i)) + elapsed := time.Since(start) + if elapsed > res.maxLatency { + res.maxLatency, res.slowest = elapsed, i + } + if err != nil { + res.errs = append(res.errs, fmt.Errorf("line %d: %w", i, err)) + } + if onLine != nil { + onLine(i) + } + } + return res +} + +func appendOnce(path, line string) error { + f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0644) + if err != nil { + return fmt.Errorf("open: %w", err) + } + if _, err = f.WriteString(line); err != nil { + f.Close() + return fmt.Errorf("write: %w", err) + } + if err = f.Close(); err != nil { + return fmt.Errorf("close: %w", err) + } + return nil +} + +// reader is the `tail -f` side: it keeps re-reading the whole file from the +// other mount and records every failure that is not "not created yet". +type reader struct { + stop chan struct{} + done chan struct{} + errs []error +} + +func startReader(path string) *reader { + r := &reader{stop: make(chan struct{}), done: make(chan struct{})} + go func() { + defer close(r.done) + for { + select { + case <-r.stop: + return + default: + } + if _, err := os.ReadFile(path); err != nil && !os.IsNotExist(err) { + r.errs = append(r.errs, err) + } + time.Sleep(200 * time.Millisecond) + } + }() + return r +} + +func (r *reader) Stop() []error { + close(r.stop) + <-r.done + return r.errs +} + +// waitForContent re-reads path until it matches want or the timeout passes. +// A mount caches metadata for about a second, so a read taken the instant the +// writer's last close returned can legitimately still be behind; content that +// is wrong rather than merely late never converges and still fails. +func waitForContent(path string, want []byte, timeout time.Duration) (got []byte, ok bool) { + deadline := time.Now().Add(timeout) + for { + data, err := os.ReadFile(path) + if err == nil { + got = data + if bytes.Equal(got, want) { + return got, true + } + } + if time.Now().After(deadline) { + return got, false + } + time.Sleep(250 * time.Millisecond) + } +} + +func expectedAppendContent(lines int) []byte { + var buf []byte + for i := 0; i < lines; i++ { + buf = append(buf, []byte(fmt.Sprintf("%07d\n", i))...) + } + return buf +} + +// TestReadWithVolumeServerDown covers the original report: a file written while +// everything was healthy must stay readable from a second mount after any +// single volume server goes away. +func TestReadWithVolumeServerDown(t *testing.T) { + c := startFailoverCluster(t, 3, 2) + + const fileSize = 8 << 20 + payloads := make(map[string][]byte) + for i := 0; i < 3; i++ { + name := fmt.Sprintf("readfile-%d", i) + data := make([]byte, fileSize) + _, err := rand.Read(data) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(c.MountDir(0), name), data, 0644)) + payloads[name] = data + } + // Let the filer commit the last chunks before anything is torn down. + time.Sleep(2 * time.Second) + + t.Logf("topology before chaos:\n%s", c.MasterGet("/dir/status?pretty=y")) + + alreadyRead := make(map[string]bool) + for victim := 0; victim < 3; victim++ { + // Placement decides which two of the three servers hold each volume, so + // pick a file this victim actually backs: reading one it never held + // would pass without exercising recovery at all. Prefer a file no + // earlier iteration has read, whose chunks the reader has not cached. + name := "" + for _, allowRead := range []bool{false, true} { + for i := 0; i < 3 && name == ""; i++ { + candidate := fmt.Sprintf("readfile-%d", i) + if alreadyRead[candidate] && !allowRead { + continue + } + onVictim, err := c.FileIsOn("/"+candidate, c.VolumeServerAddress(victim)) + require.NoError(t, err, "resolve placement of %s", candidate) + if onVictim { + name = candidate + } + } + } + require.NotEmpty(t, name, "no test file has a replica on volume%d, nothing to fail over from\n%s", + victim, c.MasterGet("/dir/status?pretty=y")) + alreadyRead[name] = true + + c.KillVolume(victim) + + start := time.Now() + got, err := os.ReadFile(filepath.Join(c.MountDir(1), name)) + elapsed := time.Since(start) + if err != nil { + t.Logf("topology with volume%d down:\n%s", victim, c.MasterGet("/dir/status?pretty=y")) + } + require.NoError(t, err, "read %s with volume %d down\n%s", name, victim, c.tailLog("mount1")) + require.Equal(t, sha256.Sum256(payloads[name]), sha256.Sum256(got), + "content mismatch for %s with volume %d down", name, victim) + t.Logf("read %s (held by volume%d) with volume%d down in %v", name, victim, victim, elapsed) + + require.NoError(t, c.StartVolume(victim)) + time.Sleep(3 * time.Second) // let the master see the heartbeat again + } +} + +// TestReadAfterCachedLocationDies pins the reason the read path needs a cache +// invalidator at all. Reading a file for the first time after a server dies +// proves nothing: the lookup is fresh and simply returns the survivor. The +// damage needs a reader whose cached location list is both stale and useless, +// which is what this sequence builds: +// +// 1. kill one of the two servers holding a volume and wait for the master to +// drop it, so a lookup now resolves to the survivor alone; +// 2. read a file on that volume, which caches exactly that one location; +// 3. bring the first server back, which the reader never hears about; +// 4. kill the survivor, and read another file on the same volume. +// +// The reader's only cached location is now dead while the data is live on the +// restarted server. Without the invalidator the mount retries the corpse until +// it gives up with EIO. +func TestReadAfterCachedLocationDies(t *testing.T) { + c := startFailoverCluster(t, 3, 2) + + // Small files so each is a single chunk on a single volume, which keeps the + // mapping from file to server unambiguous. + const fileSize = 256 << 10 + payloads := make(map[string][]byte) + for i := 0; i < 6; i++ { + name := fmt.Sprintf("smallfile-%d", i) + data := make([]byte, fileSize) + _, err := rand.Read(data) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(c.MountDir(0), name), data, 0644)) + payloads[name] = data + } + time.Sleep(2 * time.Second) + + // Two files on one volume: one to prime the reader's cache, one to probe + // with afterwards. The probe has to be a file the reader has never read, or + // its chunk would come from the local cache without a lookup at all. + filesByVolume := make(map[uint32][]string) + for name := range payloads { + vids, err := c.FileVolumeIds("/" + name) + require.NoError(t, err, "resolve volumes of %s", name) + require.Len(t, vids, 1, "%s should be a single chunk", name) + filesByVolume[vids[0]] = append(filesByVolume[vids[0]], name) + } + var vid uint32 + var prime, probe string + for candidate, names := range filesByVolume { + if len(names) >= 2 { + vid, prime, probe = candidate, names[0], names[1] + break + } + } + require.NotEmpty(t, prime, "no volume holds two of the test files: %v", filesByVolume) + + holders, err := c.VolumeHolders(vid) + require.NoError(t, err) + require.Len(t, holders, 2, "001 replication should put volume %d on two servers", vid) + first, second := c.volumeIndexOf(holders[0]), c.volumeIndexOf(holders[1]) + require.NotEqual(t, -1, first) + require.NotEqual(t, -1, second) + + c.KillVolume(first) + _, err = c.WaitForHolders(vid, 1, 90*time.Second) + require.NoError(t, err, "master did not drop the dead server") + + got, err := os.ReadFile(filepath.Join(c.MountDir(1), prime)) + require.NoError(t, err, "priming read\n%s", c.tailLog("mount1")) + require.Equal(t, sha256.Sum256(payloads[prime]), sha256.Sum256(got)) + + require.NoError(t, c.StartVolume(first)) + _, err = c.WaitForHolders(vid, 2, 90*time.Second) + require.NoError(t, err, "restarted server did not re-register") + + c.KillVolume(second) + start := time.Now() + got, err = os.ReadFile(filepath.Join(c.MountDir(1), probe)) + require.NoError(t, err, "probe read after the cached location died\n%s", c.tailLog("mount1")) + require.Equal(t, sha256.Sum256(payloads[probe]), sha256.Sum256(got)) + t.Logf("recovered %s from the restarted server in %v", probe, time.Since(start)) +} + +// TestAppendWithoutChaos is the control for the chaos runs below: the same +// append-and-tail workload with nothing being stopped or started. +func TestAppendWithoutChaos(t *testing.T) { + c := startFailoverCluster(t, 3, 2) + runChaosAppend(t, c, "no-chaos", func() {}) +} + +// TestAppendWhileVolumeServerStops is scenario "STOP volumes": append from +// mount0 and tail from mount1 while one volume server is killed mid-stream. +func TestAppendWhileVolumeServerStops(t *testing.T) { + for victim := 0; victim < 3; victim++ { + t.Run(fmt.Sprintf("volume%d", victim), func(t *testing.T) { + c := startFailoverCluster(t, 3, 2) + runChaosAppend(t, c, fmt.Sprintf("stop-%d", victim), func() { + c.KillVolume(victim) + }) + }) + } +} + +// TestAppendWhileVolumeServerStarts is scenario "Start volumes": the cluster is +// already one server down when the writer starts, and that server comes back +// mid-stream. This is where the reporter saw 12-38 s stalls per append. +func TestAppendWhileVolumeServerStarts(t *testing.T) { + for victim := 0; victim < 3; victim++ { + t.Run(fmt.Sprintf("volume%d", victim), func(t *testing.T) { + c := startFailoverCluster(t, 3, 2) + c.KillVolume(victim) + time.Sleep(3 * time.Second) + runChaosAppend(t, c, fmt.Sprintf("start-%d", victim), func() { + require.NoError(t, c.StartVolume(victim)) + }) + }) + } +} + +// TestAppendWhileVolumeServerRestarts is scenario "Re-start volumes". +func TestAppendWhileVolumeServerRestarts(t *testing.T) { + for victim := 0; victim < 3; victim++ { + t.Run(fmt.Sprintf("volume%d", victim), func(t *testing.T) { + c := startFailoverCluster(t, 3, 2) + runChaosAppend(t, c, fmt.Sprintf("restart-%d", victim), func() { + c.KillVolume(victim) + time.Sleep(2 * time.Second) + require.NoError(t, c.StartVolume(victim)) + }) + }) + } +} + +// TestLargeWriteWhileVolumeServerStops is the part-2 variant: a single large +// copy instead of many small appends, checksum-verified from the other mount. +func TestLargeWriteWhileVolumeServerStops(t *testing.T) { + c := startFailoverCluster(t, 3, 2) + + const size = 64 << 20 + data := make([]byte, size) + _, err := rand.Read(data) + require.NoError(t, err) + + path := filepath.Join(c.MountDir(0), "largefile") + f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + require.NoError(t, err) + + killed := false + const block = 1 << 20 + for off := 0; off < size; off += block { + if !killed && off >= size/3 { + c.KillVolume(2) + killed = true + } + _, werr := f.Write(data[off : off+block]) + require.NoError(t, werr, "write at offset %d with volume2 down\n%s", off, c.tailLog("mount0")) + } + require.NoError(t, f.Close()) + + got, converged := waitForContent(filepath.Join(c.MountDir(1), "largefile"), data, convergeTimeout) + require.True(t, converged, "large file did not converge on mount1: want %d bytes %x, got %d bytes %x\n%s", + len(data), sha256.Sum256(data), len(got), sha256.Sum256(got), c.tailLog("mount1")) +} + +// runChaosAppend appends from mount0 with a reader tailing on mount1, firing +// chaos once the writer is a quarter of the way in. +func runChaosAppend(t *testing.T, c *failoverCluster, name string, chaos func()) { + t.Helper() + + writePath := filepath.Join(c.MountDir(0), name) + readPath := filepath.Join(c.MountDir(1), name) + + r := startReader(readPath) + + var once sync.Once + res := appendLoop(writePath, appendLines, func(i int) { + if i == appendLines/4 { + once.Do(chaos) + } + }) + readErrs := r.Stop() + + require.Empty(t, res.errs, "append errors\n%s", c.tailLog("mount0")) + require.Empty(t, readErrs, "reader errors\n%s", c.tailLog("mount1")) + t.Logf("%s: slowest append was line %d at %v", name, res.slowest, res.maxLatency) + // A healthy append stays well under a second; the reported regression parked + // single appends at 12-38 s while a volume server was coming back. + require.Less(t, res.maxLatency, maxAppendLatency, + "append %d stalled for %v\n%s", res.slowest, res.maxLatency, c.tailLog("mount0")) + + wantBytes := expectedAppendContent(appendLines) + want := string(wantBytes) + gotBytes, converged := waitForContent(readPath, wantBytes, convergeTimeout) + got := string(gotBytes) + if converged { + return + } + + // The reader never caught up. Show where it diverges and what the writer's + // own mount and the filer make of the same file, which says whether the + // data was lost on the way in or is only invisible from this side. + d := firstDiff(want, got) + fromWriter, _ := os.ReadFile(writePath) + viaFiler, filerErr := c.FilerGet("/" + name) + require.Failf(t, "final content mismatch", + "%s: first difference at offset %d (want %d bytes, got %d)\nwant %q\ngot %q\nmount0 matches=%v filer matches=%v (err %v)\n%s", + name, d, len(want), len(got), window(want, d), window(got, d), + string(fromWriter) == want, string(viaFiler) == want, filerErr, + c.tailLog("mount0")) +} + +// firstDiff returns the offset of the first differing byte, or -1 when equal. +func firstDiff(want, got string) int { + for i := 0; i < len(want) && i < len(got); i++ { + if want[i] != got[i] { + return i + } + } + if len(want) != len(got) { + return min(len(want), len(got)) + } + return -1 +} + +func window(s string, at int) string { + return s[max(0, at-24):min(len(s), at+24)] +} diff --git a/weed/filer/filechunk_group.go b/weed/filer/filechunk_group.go index 55a4ab4eb..87c700847 100644 --- a/weed/filer/filechunk_group.go +++ b/weed/filer/filechunk_group.go @@ -26,7 +26,7 @@ type ChunkGroup struct { // - Read-ahead prefetch parallelism // - Number of concurrent section reads for large files // If concurrentReaders <= 0, defaults to 16. -func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_cache.ChunkCache, chunks []*filer_pb.FileChunk, concurrentReaders int) (*ChunkGroup, error) { +func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_cache.ChunkCache, chunks []*filer_pb.FileChunk, concurrentReaders int, cacheInvalidator CacheInvalidator) (*ChunkGroup, error) { if concurrentReaders <= 0 { concurrentReaders = 16 } @@ -41,7 +41,7 @@ func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_ group := &ChunkGroup{ lookupFn: lookupFn, sections: make(map[SectionIndex]*FileChunkSection), - readerCache: NewReaderCache(readerCacheLimit, chunkCache, lookupFn, nil), + readerCache: NewReaderCache(readerCacheLimit, chunkCache, lookupFn, cacheInvalidator), concurrentReaders: concurrentReaders, } diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 5b956aeb7..98f7b274a 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -130,7 +130,7 @@ func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) { fileSize := filer.FileSize(entry) entry.Attributes.FileSize = fileSize var resolveManifestErr error - fh.entryChunkGroup, resolveManifestErr = filer.NewChunkGroup(fh.wfs.LookupFn(), fh.wfs.chunkCache, entry.Chunks, fh.wfs.option.ConcurrentReaders) + fh.entryChunkGroup, resolveManifestErr = filer.NewChunkGroup(fh.wfs.LookupFn(), fh.wfs.chunkCache, entry.Chunks, fh.wfs.option.ConcurrentReaders, fh.wfs.CacheInvalidator()) if resolveManifestErr != nil { glog.Warningf("failed to resolve manifest chunks in %+v", entry) } diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index e4858f593..44a268896 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -962,6 +962,17 @@ func (wfs *WFS) LookupFn() wdclient.LookupFileIdFunctionType { return wfs.filerClient.GetLookupFileIdFunction() } +// CacheInvalidator lets a chunk read that failed against every cached location +// drop that entry and look the volume up again, so a mount does not keep +// hammering a volume server that has since moved or died. Nil under filerProxy, +// where there is no filerClient and LookupFn never consults a location cache. +func (wfs *WFS) CacheInvalidator() filer.CacheInvalidator { + if wfs.filerClient == nil { + return nil + } + return wfs.filerClient +} + func (wfs *WFS) getCurrentFiler() pb.ServerAddress { i := atomic.LoadInt32(&wfs.option.filerIndex) return wfs.option.FilerAddresses[i] diff --git a/weed/mount/weedfs_attr_race_test.go b/weed/mount/weedfs_attr_race_test.go index f7ced56f7..3ffbbc97b 100644 --- a/weed/mount/weedfs_attr_race_test.go +++ b/weed/mount/weedfs_attr_race_test.go @@ -34,7 +34,7 @@ func TestAttrChunkRace(t *testing.T) { Name: "sample.txt", Attributes: &filer_pb.FuseAttributes{FileMode: 0644}, } - chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1) + chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1, nil) if err != nil { t.Fatalf("NewChunkGroup: %v", err) } @@ -114,7 +114,7 @@ func TestReadFromChunksRace(t *testing.T) { Name: "sample.txt", Attributes: &filer_pb.FuseAttributes{FileMode: 0644}, } - chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1) + chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1, nil) if err != nil { t.Fatalf("NewChunkGroup: %v", err) }