diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index e7da65c35..be6554432 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -92,6 +92,11 @@ jobs: for target in windows/amd64 windows/arm64 freebsd/amd64 darwin/arm64; do echo "== $target" GOOS=${target%/*} GOARCH=${target#*/} go build ./weed/... + # Tests too: they reach for per-OS syscall constants the build does + # not, and only compiling them catches an untagged one. + GOOS=${target%/*} GOARCH=${target#*/} go vet ./weed/mount/... ./weed/command/... 2>&1 | + grep -v "MessageState contains sync.Mutex" | tee /tmp/vet-$$.txt + if grep -q "vet:" /tmp/vet-$$.txt; then exit 1; fi done test: diff --git a/.github/workflows/mount-windows.yml b/.github/workflows/mount-windows.yml new file mode 100644 index 000000000..bf2235cdc --- /dev/null +++ b/.github/workflows/mount-windows.yml @@ -0,0 +1,183 @@ +name: "mount: windows" + +on: + push: + branches: [ master ] + paths: + - 'weed/mount/**' + - 'weed/command/mount*.go' + - 'test/winfsp/**' + - '.github/workflows/mount-windows.yml' + # No base branch filter: this is the only thing that runs the Windows mount, + # so it should cover a pull request stacked on another one too. + pull_request: + paths: + - 'weed/mount/**' + - 'weed/command/mount*.go' + - 'test/winfsp/**' + - '.github/workflows/mount-windows.yml' + +concurrency: + group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + mount-windows: + name: Mount on Windows + runs-on: windows-latest + timeout-minutes: 40 + env: + # The runner ships MinGW, so cgo is on by default and cgofuse picks its + # cgo variant, which wants WinFsp's headers. The nocgo variant loads + # winfsp-x64.dll at run time instead, which is how weed.exe is released. + CGO_ENABLED: 0 + + steps: + - uses: actions/checkout@v7 + with: + persist-credentials: false + + - uses: actions/setup-go@v7 + with: + go-version-file: 'go.mod' + + # cgofuse loads winfsp-x64.dll at run time, so WinFsp is needed here but + # not to build. + - name: Install WinFsp + run: choco install winfsp -y --no-progress + + - name: Build weed.exe + run: go build -o weed.exe ./weed + + # The runner tears down a step's process tree when its shell exits, so a + # cluster started in one step is gone by the next. Everything that needs + # the cluster and the mount alive has to share a step. + - name: Mount and exercise + shell: pwsh + run: | + $ErrorActionPreference = 'Stop' + + function Test-Port($port) { + # A plain connect, because Test-NetConnection has reported success + # here for a port nothing was listening on. + $client = New-Object System.Net.Sockets.TcpClient + try { $client.Connect('127.0.0.1', $port); return $client.Connected } + catch { return $false } + finally { $client.Dispose() } + } + + function Start-Mount($log) { + Start-Process -FilePath .\weed.exe ` + -ArgumentList '-logtostderr','mount','-filer=127.0.0.1:8888','-dir=S:' ` + -RedirectStandardOutput "C:\$log.log" -RedirectStandardError "C:\$log.err.log" + $deadline = (Get-Date).AddMinutes(2) + while ((Get-Date) -lt $deadline) { + if (Test-Path S:\) { Write-Host "S: is mounted"; return } + Start-Sleep -Seconds 2 + } + Get-Content "C:\$log.log", "C:\$log.err.log" -ErrorAction SilentlyContinue + throw "S: never appeared" + } + + function Stop-Mount { + Get-CimInstance Win32_Process -Filter "Name = 'weed.exe'" | + Where-Object { $_.CommandLine -like '*mount*' } | + ForEach-Object { Stop-Process -Id $_.ProcessId -Force } + $deadline = (Get-Date).AddMinutes(1) + while ((Get-Date) -lt $deadline -and (Test-Path S:\)) { Start-Sleep -Seconds 2 } + if (Test-Path S:\) { throw "S: still present after stopping the mount" } + Write-Host "unmounted" + } + + function Invoke-Tests($label, [string[]]$goArgs) { + Write-Host "::group::$label" + & go @goArgs + $code = $LASTEXITCODE + Write-Host "::endgroup::" + if ($code -ne 0) { throw "$label failed with exit $code" } + } + + New-Item -ItemType Directory -Force -Path C:\seaweed-data | Out-Null + # -ip pins the cluster to loopback; it otherwise advertises and binds + # the runner's LAN address, which 127.0.0.1 cannot reach. + Start-Process -FilePath .\weed.exe ` + -ArgumentList '-logtostderr','mini','-dir=C:\seaweed-data','-ip=127.0.0.1' ` + -RedirectStandardOutput C:\seaweed-mini.log -RedirectStandardError C:\seaweed-mini.err.log + + $deadline = (Get-Date).AddMinutes(3) + while ((Get-Date) -lt $deadline) { + # The mount dials grpc, not http, so both ports have to answer. + if ((Test-Port 8888) -and (Test-Port 18888)) { break } + Start-Sleep -Seconds 3 + } + if (-not ((Test-Port 8888) -and (Test-Port 18888))) { + Get-Content C:\seaweed-mini.log, C:\seaweed-mini.err.log -ErrorAction SilentlyContinue + throw "filer never came up" + } + Write-Host "filer is up on http 8888 and grpc 18888" + + Start-Mount 'seaweed-mount' + + Invoke-Tests 'exercise' @('test','-v','-timeout','20m','./test/winfsp','-mountpoint=S:\') + Invoke-Tests 'persist-write' @('test','-v','-timeout','15m','./test/winfsp','-run','TestPersistence','-mountpoint=S:\','-phase=write','-filer=127.0.0.1:8888') + + Stop-Mount + Start-Mount 'seaweed-remount' + + Invoke-Tests 'persist-verify' @('test','-v','-timeout','15m','./test/winfsp','-run','TestPersistence','-mountpoint=S:\','-phase=verify') + + # WinFsp creates the mount directory itself, so the path must not + # exist; only its parent has to. + Write-Host "::group::mount over a directory" + Stop-Mount + Remove-Item C:\seaweed-mnt -Recurse -Force -ErrorAction SilentlyContinue + Start-Process -FilePath .\weed.exe ` + -ArgumentList '-logtostderr','mount','-filer=127.0.0.1:8888','-dir=C:\seaweed-mnt' ` + -RedirectStandardOutput C:\seaweed-dirmount.log -RedirectStandardError C:\seaweed-dirmount.err.log + $deadline = (Get-Date).AddMinutes(2) + $ok = $false + while ((Get-Date) -lt $deadline) { + # Listing succeeds on the plain empty directory too, so wait for the + # reparse point WinFsp turns it into. Otherwise this step passes + # without a mount and writes to local disk. + $item = Get-Item C:\seaweed-mnt -Force -ErrorAction SilentlyContinue + if ($null -ne $item -and $item.Attributes.ToString() -like '*ReparsePoint*') { $ok = $true; break } + Start-Sleep -Seconds 2 + } + if (-not $ok) { + Get-Content C:\seaweed-dirmount.log, C:\seaweed-dirmount.err.log -ErrorAction SilentlyContinue + throw "mounting over a directory failed" + } + Set-Content -Path C:\seaweed-mnt\dirmount.txt -Value 'via directory mount' + if ((Get-Content C:\seaweed-mnt\dirmount.txt) -ne 'via directory mount') { throw "readback through the directory mount differs" } + Remove-Item C:\seaweed-mnt\dirmount.txt -Force + Get-CimInstance Win32_Process -Filter "Name = 'weed.exe'" | + Where-Object { $_.CommandLine -like '*mount*' } | + ForEach-Object { Stop-Process -Id $_.ProcessId -Force } + Start-Sleep -Seconds 5 + Write-Host "::endgroup::" + + Start-Mount 'seaweed-remount2' + + Write-Host "::group::explorer-style walk" + New-Item -ItemType Directory -Force -Path S:\walk | Out-Null + 1..200 | ForEach-Object { Set-Content -Path "S:\walk\f$_.txt" -Value "line $_" } + $count = (Get-ChildItem S:\walk | Measure-Object).Count + if ($count -ne 200) { throw "listed $count files, expected 200" } + $body = Get-Content S:\walk\f42.txt + if ($body -ne 'line 42') { throw "unexpected content: $body" } + Copy-Item S:\walk\f42.txt S:\walk\copy.txt + Remove-Item S:\walk -Recurse -Force + if (Test-Path S:\walk) { throw "directory survived recursive delete" } + Write-Host "::endgroup::" + + - name: Logs + if: always() + shell: pwsh + run: | + foreach ($f in 'C:\seaweed-mount.log','C:\seaweed-mount.err.log','C:\seaweed-remount.log','C:\seaweed-remount.err.log','C:\seaweed-remount2.log','C:\seaweed-remount2.err.log','C:\seaweed-dirmount.log','C:\seaweed-dirmount.err.log','C:\seaweed-mini.log','C:\seaweed-mini.err.log') { + if (Test-Path $f) { Write-Host "===== $f"; Get-Content $f -Tail 200 } + } diff --git a/go.mod b/go.mod index 4f55585f6..da5a01a30 100644 --- a/go.mod +++ b/go.mod @@ -471,6 +471,7 @@ require ( github.com/unknwon/goconfig v1.0.0 // indirect github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect + github.com/winfsp/cgofuse v1.6.1-0.20260126094232-f2c4fccdb286 github.com/xanzy/ssh-agent v0.3.3 // indirect github.com/yandex-cloud/go-genproto v0.0.0-20211115083454-9ca41db5ed9e // indirect github.com/ydb-platform/ydb-go-genproto v0.0.0-20260428144813-1c07baab7f7b // indirect diff --git a/go.sum b/go.sum index 3db49517a..6139db1a5 100644 --- a/go.sum +++ b/go.sum @@ -1998,6 +1998,8 @@ github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IU github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok= github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g= github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds= +github.com/winfsp/cgofuse v1.6.1-0.20260126094232-f2c4fccdb286 h1:tw5GqRXqExB/xghPoPLtVujBe9w9Pg1G78tvXCJNJAA= +github.com/winfsp/cgofuse v1.6.1-0.20260126094232-f2c4fccdb286/go.mod h1:uxjoF2jEYT3+x+vC2KJddEGdk/LU8pRowXmyVMHSV5I= github.com/wk8/go-ordered-map/v2 v2.1.8 h1:5h/BUHu93oj4gIdvHHHGsScSTMijfx5PeYkE/fJgbpc= github.com/wk8/go-ordered-map/v2 v2.1.8/go.mod h1:5nJHM5DyteebpVlHnWMV0rPz6Zp7+xBAnxjb1X5vnTw= github.com/wsxiaoys/terminal v0.0.0-20160513160801-0940f3fc43a0 h1:3UeQBvD0TFrlVjOeLOBz+CPAI8dnbqNSVwUwRrkp7vQ= diff --git a/test/winfsp/diskfree_other.go b/test/winfsp/diskfree_other.go new file mode 100644 index 000000000..d4a1d3399 --- /dev/null +++ b/test/winfsp/diskfree_other.go @@ -0,0 +1,9 @@ +//go:build !windows + +package winfsp + +import "errors" + +func getDiskFreeSpace(path string, free, total, totalFree *uint64) error { + return errors.New("windows only") +} diff --git a/test/winfsp/diskfree_windows.go b/test/winfsp/diskfree_windows.go new file mode 100644 index 000000000..9a4f71f89 --- /dev/null +++ b/test/winfsp/diskfree_windows.go @@ -0,0 +1,13 @@ +package winfsp + +import ( + "golang.org/x/sys/windows" +) + +func getDiskFreeSpace(path string, free, total, totalFree *uint64) error { + p, err := windows.UTF16PtrFromString(path) + if err != nil { + return err + } + return windows.GetDiskFreeSpaceEx(p, free, total, totalFree) +} diff --git a/test/winfsp/mount_test.go b/test/winfsp/mount_test.go new file mode 100644 index 000000000..9ab5764d7 --- /dev/null +++ b/test/winfsp/mount_test.go @@ -0,0 +1,251 @@ +// Package winfsp exercises a live SeaweedFS mount on Windows. +// +// It runs against whatever is mounted at -mountpoint, so it needs a real +// WinFsp mount and is skipped otherwise. The CI job drives it; run it by hand +// with: go test ./test/winfsp -mountpoint=S:\ +package winfsp + +import ( + "bytes" + "crypto/rand" + "flag" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "testing" +) + +var mountPoint = flag.String("mountpoint", "", "a mounted SeaweedFS drive, e.g. S:\\") + +// entryCount is the directory width the test builds. The reported case that +// motivated Windows support is 200k files in one folder; CI uses less so the +// job stays inside its budget, but the read path is the same. +var entryCount = flag.Int("entries", 5000, "how many files to put in one directory") + +func testRoot(t *testing.T) string { + t.Helper() + if *mountPoint == "" { + t.Skip("no -mountpoint given; this test needs a live WinFsp mount") + } + dir := filepath.Join(*mountPoint, "winfsp-test-"+t.Name()) + if err := os.RemoveAll(dir); err != nil && !os.IsNotExist(err) { + t.Fatalf("clean %s: %v", dir, err) + } + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatalf("mkdir %s: %v", dir, err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + return dir +} + +func TestWriteReadRoundTrip(t *testing.T) { + dir := testRoot(t) + for _, size := range []int{0, 1, 4095, 4096, 1 << 20, 5 << 20} { + t.Run(fmt.Sprintf("%dbytes", size), func(t *testing.T) { + want := make([]byte, size) + if _, err := rand.Read(want); err != nil { + t.Fatalf("rand: %v", err) + } + path := filepath.Join(dir, fmt.Sprintf("file-%d", size)) + if err := os.WriteFile(path, want, 0644); err != nil { + t.Fatalf("write: %v", err) + } + got, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("content differs: got %d bytes, want %d", len(got), len(want)) + } + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat: %v", err) + } + if info.Size() != int64(size) { + t.Fatalf("stat size = %d, want %d", info.Size(), size) + } + }) + } +} + +func TestSeekWriteAndOverwrite(t *testing.T) { + dir := testRoot(t) + path := filepath.Join(dir, "sparse") + + f, err := os.Create(path) + if err != nil { + t.Fatalf("create: %v", err) + } + if _, err := f.WriteAt([]byte("tail"), 4096); err != nil { + t.Fatalf("write at offset: %v", err) + } + if _, err := f.WriteAt([]byte("head"), 0); err != nil { + t.Fatalf("write at start: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + got, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if len(got) != 4100 { + t.Fatalf("size = %d, want 4100", len(got)) + } + if string(got[:4]) != "head" || string(got[4096:]) != "tail" { + t.Fatalf("content at the edges is wrong: %q ... %q", got[:4], got[4096:]) + } + if !bytes.Equal(got[4:4096], make([]byte, 4092)) { + t.Fatal("the hole between the two writes is not zero-filled") + } +} + +func TestRenameAndDelete(t *testing.T) { + dir := testRoot(t) + src := filepath.Join(dir, "before") + dst := filepath.Join(dir, "after") + + if err := os.WriteFile(src, []byte("payload"), 0644); err != nil { + t.Fatalf("write: %v", err) + } + if err := os.Rename(src, dst); err != nil { + t.Fatalf("rename: %v", err) + } + if _, err := os.Stat(src); !os.IsNotExist(err) { + t.Fatalf("old name still resolves: %v", err) + } + got, err := os.ReadFile(dst) + if err != nil { + t.Fatalf("read renamed: %v", err) + } + if string(got) != "payload" { + t.Fatalf("content after rename = %q", got) + } + if err := os.Remove(dst); err != nil { + t.Fatalf("remove: %v", err) + } + if _, err := os.Stat(dst); !os.IsNotExist(err) { + t.Fatalf("deleted file still resolves: %v", err) + } +} + +func TestDirectoryTree(t *testing.T) { + dir := testRoot(t) + nested := filepath.Join(dir, "a", "b", "c") + if err := os.MkdirAll(nested, 0755); err != nil { + t.Fatalf("mkdirall: %v", err) + } + if err := os.WriteFile(filepath.Join(nested, "leaf"), []byte("x"), 0644); err != nil { + t.Fatalf("write leaf: %v", err) + } + info, err := os.Stat(nested) + if err != nil { + t.Fatalf("stat dir: %v", err) + } + if !info.IsDir() { + t.Fatal("nested path is not reported as a directory") + } + // A non-empty directory must not be removable. + if err := os.Remove(filepath.Join(dir, "a", "b")); err == nil { + t.Fatal("removed a non-empty directory") + } + if err := os.RemoveAll(filepath.Join(dir, "a")); err != nil { + t.Fatalf("removeall: %v", err) + } +} + +// TestWideDirectory is the case from the report: one folder holding far more +// entries than Explorer or the Windows WebDAV client cope with. +func TestWideDirectory(t *testing.T) { + dir := testRoot(t) + want := make([]string, 0, *entryCount) + for i := 0; i < *entryCount; i++ { + name := fmt.Sprintf("img-%06d.dat", i) + if err := os.WriteFile(filepath.Join(dir, name), []byte{byte(i)}, 0644); err != nil { + t.Fatalf("write %s: %v", name, err) + } + want = append(want, name) + } + + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("readdir: %v", err) + } + got := make([]string, 0, len(entries)) + for _, e := range entries { + got = append(got, e.Name()) + } + sort.Strings(got) + sort.Strings(want) + if len(got) != len(want) { + t.Fatalf("readdir returned %d entries, want %d", len(got), len(want)) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("entry %d = %q, want %q", i, got[i], want[i]) + } + } + + // Reading the directory twice must be stable; a paging bug shows up here. + again, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("second readdir: %v", err) + } + if len(again) != len(entries) { + t.Fatalf("second readdir returned %d entries, first returned %d", len(again), len(entries)) + } +} + +func TestConcurrentWriters(t *testing.T) { + dir := testRoot(t) + const writers = 8 + const perWriter = 32 + + var wg sync.WaitGroup + errs := make(chan error, writers) + for w := 0; w < writers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for i := 0; i < perWriter; i++ { + name := filepath.Join(dir, fmt.Sprintf("w%d-%d", w, i)) + body := []byte(strings.Repeat(fmt.Sprintf("%d", w), 128)) + if err := os.WriteFile(name, body, 0644); err != nil { + errs <- fmt.Errorf("writer %d: %w", w, err) + return + } + } + }(w) + } + wg.Wait() + close(errs) + for err := range errs { + t.Fatalf("%v", err) + } + + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("readdir: %v", err) + } + if len(entries) != writers*perWriter { + t.Fatalf("got %d files, want %d", len(entries), writers*perWriter) + } +} + +func TestStatfsReportsCapacity(t *testing.T) { + if *mountPoint == "" { + t.Skip("no -mountpoint given; this test needs a live WinFsp mount") + } + // The drive has to report a size, or Explorer refuses to show it. + var free, total, totalFree uint64 + if err := getDiskFreeSpace(*mountPoint, &free, &total, &totalFree); err != nil { + t.Fatalf("GetDiskFreeSpaceEx: %v", err) + } + if total == 0 { + t.Fatal("drive reports zero total bytes") + } +} diff --git a/test/winfsp/persistence_test.go b/test/winfsp/persistence_test.go new file mode 100644 index 000000000..110fbd1ec --- /dev/null +++ b/test/winfsp/persistence_test.go @@ -0,0 +1,186 @@ +package winfsp + +import ( + "bytes" + "flag" + "fmt" + "hash/fnv" + "io" + "net/http" + "net/url" + "os" + "path" + "path/filepath" + "testing" + "time" +) + +var ( + // phase splits the test across a remount: the write phase is run, the + // mount is torn down and brought back, then the verify phase runs. Reading + // back through the same live mount proves nothing about durability, since + // the answer can come from the mount's own caches. + phase = flag.String("phase", "", "write or verify; empty skips the persistence test") + + // filerAddr enables a second check, that the bytes reached the filer and + // are servable without the mount in the path at all. + filerAddr = flag.String("filer", "", "filer host:port to cross-check through, e.g. localhost:8888") + + // persistSubdir is where the fixtures live, under both the mount and the + // filer's mount root. + persistSubdir = flag.String("persistdir", "winfsp-persist", "directory under the mount to persist into") +) + +type fixture struct { + relPath string + size int +} + +// Sizes straddle the boundaries where the write path changes behavior: inline, +// a single chunk, and several chunks. +var fixtures = []fixture{ + {"small.txt", 11}, + {"medium.bin", 300 << 10}, + {"multichunk.bin", 9 << 20}, + {"nested/deep/leaf.bin", 65536}, + {"unicode-café-日本.txt", 64}, +} + +// contentFor derives bytes from the name, so the write and verify phases agree +// without carrying a manifest between them. +func contentFor(relPath string, size int) []byte { + h := fnv.New64a() + h.Write([]byte(relPath)) + state := h.Sum64() | 1 + out := make([]byte, size) + for i := range out { + state ^= state << 13 + state ^= state >> 7 + state ^= state << 17 + out[i] = byte(state) + } + return out +} + +func TestPersistence(t *testing.T) { + if *mountPoint == "" { + t.Skip("no -mountpoint given; this test needs a live WinFsp mount") + } + switch *phase { + case "write": + persistenceWrite(t) + case "verify": + persistenceVerify(t) + default: + t.Skip("no -phase given; run with -phase=write before a remount and -phase=verify after") + } +} + +func persistenceWrite(t *testing.T) { + root := filepath.Join(*mountPoint, *persistSubdir) + if err := os.RemoveAll(root); err != nil && !os.IsNotExist(err) { + t.Fatalf("clean %s: %v", root, err) + } + if err := os.MkdirAll(root, 0755); err != nil { + t.Fatalf("mkdir %s: %v", root, err) + } + + for _, f := range fixtures { + target := filepath.Join(root, filepath.FromSlash(f.relPath)) + if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil { + t.Fatalf("mkdir for %s: %v", f.relPath, err) + } + // Sync before closing: the mount is killed rather than unmounted, so + // anything still queued for flush is legitimately lost. Testing + // durability means testing what survives after an explicit sync. + if err := writeAndSync(target, contentFor(f.relPath, f.size)); err != nil { + t.Fatalf("write %s: %v", f.relPath, err) + } + t.Logf("wrote %s (%d bytes)", f.relPath, f.size) + } + + // Closing the files should have pushed them to the filer. Check that + // directly, so a failure here separates "never left the mount" from + // "did not survive the remount". + if *filerAddr == "" { + return + } + for _, f := range fixtures { + want := contentFor(f.relPath, f.size) + got, err := fetchFromFiler(*filerAddr, path.Join(*persistSubdir, f.relPath)) + if err != nil { + t.Errorf("fetch %s from filer: %v", f.relPath, err) + continue + } + if !bytes.Equal(got, want) { + t.Errorf("filer served %d bytes for %s, want %d", len(got), f.relPath, len(want)) + } + } +} + +func persistenceVerify(t *testing.T) { + root := filepath.Join(*mountPoint, *persistSubdir) + if _, err := os.Stat(root); err != nil { + t.Fatalf("stat %s after remount: %v", root, err) + } + + for _, f := range fixtures { + target := filepath.Join(root, filepath.FromSlash(f.relPath)) + got, err := os.ReadFile(target) + if err != nil { + t.Errorf("read %s after remount: %v", f.relPath, err) + continue + } + want := contentFor(f.relPath, f.size) + if len(got) != len(want) { + t.Errorf("%s is %d bytes after remount, want %d", f.relPath, len(got), len(want)) + continue + } + if !bytes.Equal(got, want) { + t.Errorf("%s survived the remount with different content", f.relPath) + } + } + + // The directory structure has to come back too, not just the files. + entries, err := os.ReadDir(root) + if err != nil { + t.Fatalf("readdir %s after remount: %v", root, err) + } + if len(entries) == 0 { + t.Fatal("mount root directory is empty after remount") + } + + if err := os.RemoveAll(root); err != nil { + t.Errorf("cleanup: %v", err) + } +} + +func writeAndSync(path string, content []byte) error { + f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) + if err != nil { + return err + } + if _, err := f.Write(content); err != nil { + f.Close() + return err + } + if err := f.Sync(); err != nil { + f.Close() + return err + } + return f.Close() +} + +func fetchFromFiler(addr, filerPath string) ([]byte, error) { + endpoint := &url.URL{Scheme: "http", Host: addr, Path: "/" + filerPath} + client := &http.Client{Timeout: 60 * time.Second} + resp, err := client.Get(endpoint.String()) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("GET %s: %s", endpoint, resp.Status) + } + return io.ReadAll(resp.Body) +} diff --git a/test/winfsp/semantics_test.go b/test/winfsp/semantics_test.go new file mode 100644 index 000000000..3260bdbb7 --- /dev/null +++ b/test/winfsp/semantics_test.go @@ -0,0 +1,492 @@ +package winfsp + +import ( + "bytes" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +func TestTruncate(t *testing.T) { + dir := testRoot(t) + path := filepath.Join(dir, "resize") + original := bytes.Repeat([]byte("abcd"), 4096) // 16 KiB + + if err := os.WriteFile(path, original, 0644); err != nil { + t.Fatalf("write: %v", err) + } + + // Shrink. + if err := os.Truncate(path, 100); err != nil { + t.Fatalf("truncate down: %v", err) + } + got, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read after shrink: %v", err) + } + if len(got) != 100 { + t.Fatalf("size after shrink = %d, want 100", len(got)) + } + if !bytes.Equal(got, original[:100]) { + t.Fatal("the surviving prefix does not match the original") + } + + // Grow: the new space has to read back as zeros, not stale bytes. + if err := os.Truncate(path, 8192); err != nil { + t.Fatalf("truncate up: %v", err) + } + got, err = os.ReadFile(path) + if err != nil { + t.Fatalf("read after grow: %v", err) + } + if len(got) != 8192 { + t.Fatalf("size after grow = %d, want 8192", len(got)) + } + if !bytes.Equal(got[100:], make([]byte, 8192-100)) { + t.Fatal("extension is not zero-filled") + } + + // Truncate to empty. + if err := os.Truncate(path, 0); err != nil { + t.Fatalf("truncate to zero: %v", err) + } + if info, err := os.Stat(path); err != nil || info.Size() != 0 { + t.Fatalf("stat after zero truncate: size=%v err=%v", info, err) + } +} + +func TestAppend(t *testing.T) { + dir := testRoot(t) + path := filepath.Join(dir, "log") + + for i := 0; i < 5; i++ { + f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + if err != nil { + t.Fatalf("open for append: %v", err) + } + if _, err := fmt.Fprintf(f, "line %d\n", i); err != nil { + t.Fatalf("append: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("close: %v", err) + } + } + + got, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + want := "line 0\nline 1\nline 2\nline 3\nline 4\n" + if string(got) != want { + t.Fatalf("appended content = %q, want %q", got, want) + } +} + +func TestReadPastEndOfFile(t *testing.T) { + dir := testRoot(t) + path := filepath.Join(dir, "short") + if err := os.WriteFile(path, []byte("12345"), 0644); err != nil { + t.Fatalf("write: %v", err) + } + + f, err := os.Open(path) + if err != nil { + t.Fatalf("open: %v", err) + } + defer f.Close() + + buf := make([]byte, 16) + n, err := f.ReadAt(buf, 100) + if n != 0 { + t.Fatalf("read %d bytes past EOF, want 0", n) + } + if err == nil { + t.Fatal("reading past EOF returned no error, want io.EOF") + } +} + +func TestErrorPaths(t *testing.T) { + dir := testRoot(t) + + t.Run("open missing file", func(t *testing.T) { + _, err := os.Open(filepath.Join(dir, "does-not-exist")) + if !os.IsNotExist(err) { + t.Fatalf("got %v, want a not-exist error", err) + } + }) + + t.Run("exclusive create over existing", func(t *testing.T) { + path := filepath.Join(dir, "taken") + if err := os.WriteFile(path, []byte("x"), 0644); err != nil { + t.Fatalf("write: %v", err) + } + _, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0644) + if !os.IsExist(err) { + t.Fatalf("got %v, want an already-exists error", err) + } + }) + + t.Run("remove missing file", func(t *testing.T) { + if err := os.Remove(filepath.Join(dir, "never-there")); !os.IsNotExist(err) { + t.Fatalf("got %v, want a not-exist error", err) + } + }) + + t.Run("mkdir under a file", func(t *testing.T) { + path := filepath.Join(dir, "regular") + if err := os.WriteFile(path, []byte("x"), 0644); err != nil { + t.Fatalf("write: %v", err) + } + if err := os.Mkdir(filepath.Join(path, "child"), 0755); err == nil { + t.Fatal("created a directory underneath a regular file") + } + }) + + t.Run("readdir of a missing directory", func(t *testing.T) { + if _, err := os.ReadDir(filepath.Join(dir, "no-such-dir")); err == nil { + t.Fatal("listed a directory that does not exist") + } + }) +} + +func TestRenameOverExisting(t *testing.T) { + dir := testRoot(t) + src := filepath.Join(dir, "src") + dst := filepath.Join(dir, "dst") + + if err := os.WriteFile(src, []byte("new"), 0644); err != nil { + t.Fatalf("write src: %v", err) + } + if err := os.WriteFile(dst, []byte("old"), 0644); err != nil { + t.Fatalf("write dst: %v", err) + } + if err := os.Rename(src, dst); err != nil { + t.Fatalf("rename over existing: %v", err) + } + got, err := os.ReadFile(dst) + if err != nil { + t.Fatalf("read: %v", err) + } + if string(got) != "new" { + t.Fatalf("target holds %q, want the source content", got) + } + if _, err := os.Stat(src); !os.IsNotExist(err) { + t.Fatal("source survived the rename") + } +} + +func TestRenameAcrossDirectories(t *testing.T) { + dir := testRoot(t) + from := filepath.Join(dir, "from") + to := filepath.Join(dir, "to") + for _, d := range []string{from, to} { + if err := os.Mkdir(d, 0755); err != nil { + t.Fatalf("mkdir %s: %v", d, err) + } + } + + src := filepath.Join(from, "file") + dst := filepath.Join(to, "file") + if err := os.WriteFile(src, []byte("moved"), 0644); err != nil { + t.Fatalf("write: %v", err) + } + if err := os.Rename(src, dst); err != nil { + t.Fatalf("rename across directories: %v", err) + } + got, err := os.ReadFile(dst) + if err != nil { + t.Fatalf("read: %v", err) + } + if string(got) != "moved" { + t.Fatalf("content after move = %q", got) + } + + // A directory should move too, children and all. + nested := filepath.Join(from, "sub") + if err := os.Mkdir(nested, 0755); err != nil { + t.Fatalf("mkdir sub: %v", err) + } + if err := os.WriteFile(filepath.Join(nested, "child"), []byte("c"), 0644); err != nil { + t.Fatalf("write child: %v", err) + } + if err := os.Rename(nested, filepath.Join(to, "sub")); err != nil { + t.Fatalf("rename directory: %v", err) + } + if _, err := os.Stat(filepath.Join(to, "sub", "child")); err != nil { + t.Fatalf("child did not come along: %v", err) + } +} + +// TestAwkwardNames covers what has to survive the UTF-16 boundary and the +// shapes Windows software tends to produce. +func TestAwkwardNames(t *testing.T) { + dir := testRoot(t) + names := []string{ + "café.txt", + "日本語のファイル.dat", + "emoji-🐟.bin", + "with space.txt", + "with.many.dots.txt", + "UPPER and lower.TXT", + "dash-and_underscore.txt", + "'quoted'.txt", + "(parens).txt", + "#hash&.txt", + strings.Repeat("x", 200) + ".txt", + } + for _, name := range names { + t.Run(name, func(t *testing.T) { + path := filepath.Join(dir, name) + want := []byte(name) + if err := os.WriteFile(path, want, 0644); err != nil { + t.Fatalf("write: %v", err) + } + got, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("content = %q, want %q", got, want) + } + }) + } + + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("readdir: %v", err) + } + found := make(map[string]bool, len(entries)) + for _, e := range entries { + found[e.Name()] = true + } + for _, name := range names { + if !found[name] { + t.Errorf("%q did not come back from readdir", name) + } + } +} + +// Windows enumerates a directory without "." and "..", and shows whatever the +// filesystem reports, so they must not reach it. os.ReadDir filters them, so +// this reads the handle the way Windows tooling does. +func TestNoDotEntriesInListing(t *testing.T) { + dir := testRoot(t) + for i := 0; i < 3; i++ { + if err := os.WriteFile(filepath.Join(dir, fmt.Sprintf("f%d", i)), []byte("x"), 0644); err != nil { + t.Fatalf("write: %v", err) + } + } + f, err := os.Open(dir) + if err != nil { + t.Fatalf("open dir: %v", err) + } + defer f.Close() + names, err := f.Readdirnames(-1) + if err != nil { + t.Fatalf("readdirnames: %v", err) + } + for _, name := range names { + if name == "." || name == ".." { + t.Errorf("listing includes %q", name) + } + } + if len(names) != 3 { + t.Fatalf("listing has %d entries (%v), want 3", len(names), names) + } +} + +func TestDeepDirectoryNesting(t *testing.T) { + dir := testRoot(t) + deep := dir + for i := 0; i < 24; i++ { + deep = filepath.Join(deep, fmt.Sprintf("level%02d", i)) + } + if err := os.MkdirAll(deep, 0755); err != nil { + t.Fatalf("mkdirall depth 24: %v", err) + } + leaf := filepath.Join(deep, "leaf.txt") + if err := os.WriteFile(leaf, []byte("bottom"), 0644); err != nil { + t.Fatalf("write leaf: %v", err) + } + got, err := os.ReadFile(leaf) + if err != nil { + t.Fatalf("read leaf: %v", err) + } + if string(got) != "bottom" { + t.Fatalf("leaf content = %q", got) + } +} + +// TestConcurrentSameFile is the interleaving that matters: several handles on +// one file at once, rather than one writer each on separate files. +func TestConcurrentSameFile(t *testing.T) { + dir := testRoot(t) + path := filepath.Join(dir, "shared") + + const writers = 4 + const blockSize = 4096 + if err := os.WriteFile(path, make([]byte, writers*blockSize), 0644); err != nil { + t.Fatalf("preallocate: %v", err) + } + + var wg sync.WaitGroup + errs := make(chan error, writers) + for w := 0; w < writers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + f, err := os.OpenFile(path, os.O_WRONLY, 0644) + if err != nil { + errs <- fmt.Errorf("writer %d open: %w", w, err) + return + } + defer f.Close() + block := bytes.Repeat([]byte{byte('A' + w)}, blockSize) + if _, err := f.WriteAt(block, int64(w*blockSize)); err != nil { + errs <- fmt.Errorf("writer %d write: %w", w, err) + } + }(w) + } + wg.Wait() + close(errs) + for err := range errs { + t.Fatalf("%v", err) + } + + got, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if len(got) != writers*blockSize { + t.Fatalf("size = %d, want %d", len(got), writers*blockSize) + } + for w := 0; w < writers; w++ { + want := bytes.Repeat([]byte{byte('A' + w)}, blockSize) + if !bytes.Equal(got[w*blockSize:(w+1)*blockSize], want) { + t.Fatalf("block %d was not written whole", w) + } + } +} + +func TestConcurrentReaders(t *testing.T) { + dir := testRoot(t) + path := filepath.Join(dir, "read-me") + want := contentFor("concurrent-readers", 1<<20) + if err := os.WriteFile(path, want, 0644); err != nil { + t.Fatalf("write: %v", err) + } + + var wg sync.WaitGroup + errs := make(chan error, 8) + for r := 0; r < 8; r++ { + wg.Add(1) + go func() { + defer wg.Done() + got, err := os.ReadFile(path) + if err != nil { + errs <- err + return + } + if !bytes.Equal(got, want) { + errs <- fmt.Errorf("reader saw %d bytes, want %d", len(got), len(want)) + } + }() + } + wg.Wait() + close(errs) + for err := range errs { + t.Fatalf("%v", err) + } +} + +// Hard links are not something WinFsp offers, so the mount has to refuse them +// rather than appear to succeed. +func TestHardLinkUnsupported(t *testing.T) { + dir := testRoot(t) + target := filepath.Join(dir, "original") + if err := os.WriteFile(target, []byte("x"), 0644); err != nil { + t.Fatalf("write: %v", err) + } + if err := os.Link(target, filepath.Join(dir, "hardlink")); err == nil { + t.Fatal("hard link creation reported success") + } +} + +// Symlinks are refused rather than half-supported: WinFsp needs a reparse +// point to follow one, and an entry it cannot follow reads back empty. +func TestSymlinkUnsupported(t *testing.T) { + dir := testRoot(t) + target := filepath.Join(dir, "target.txt") + if err := os.WriteFile(target, []byte("pointed at"), 0644); err != nil { + t.Fatalf("write: %v", err) + } + if err := os.Symlink(target, filepath.Join(dir, "link.txt")); err == nil { + t.Fatal("symlink creation reported success") + } +} + +func TestModTimeAdvances(t *testing.T) { + dir := testRoot(t) + path := filepath.Join(dir, "touched") + if err := os.WriteFile(path, []byte("one"), 0644); err != nil { + t.Fatalf("write: %v", err) + } + first, err := os.Stat(path) + if err != nil { + t.Fatalf("stat: %v", err) + } + + time.Sleep(1100 * time.Millisecond) + if err := os.WriteFile(path, []byte("two"), 0644); err != nil { + t.Fatalf("rewrite: %v", err) + } + second, err := os.Stat(path) + if err != nil { + t.Fatalf("stat again: %v", err) + } + if !second.ModTime().After(first.ModTime()) { + t.Fatalf("mtime did not advance: %v then %v", first.ModTime(), second.ModTime()) + } +} + +func TestChtimes(t *testing.T) { + dir := testRoot(t) + path := filepath.Join(dir, "dated") + if err := os.WriteFile(path, []byte("x"), 0644); err != nil { + t.Fatalf("write: %v", err) + } + want := time.Date(2020, 3, 4, 5, 6, 7, 0, time.UTC) + if err := os.Chtimes(path, want, want); err != nil { + t.Fatalf("chtimes: %v", err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat: %v", err) + } + if diff := info.ModTime().UTC().Sub(want); diff > 2*time.Second || diff < -2*time.Second { + t.Fatalf("mtime = %v, want about %v", info.ModTime().UTC(), want) + } +} + +func TestStatfsIsSelfConsistent(t *testing.T) { + if *mountPoint == "" { + t.Skip("no -mountpoint given; this test needs a live WinFsp mount") + } + var free, total, totalFree uint64 + if err := getDiskFreeSpace(*mountPoint, &free, &total, &totalFree); err != nil { + t.Fatalf("GetDiskFreeSpaceEx: %v", err) + } + if total == 0 { + t.Fatal("drive reports zero total bytes") + } + if free > total { + t.Fatalf("free (%d) exceeds total (%d)", free, total) + } + if totalFree > total { + t.Fatalf("total free (%d) exceeds total (%d)", totalFree, total) + } +} diff --git a/weed/command/mount.go b/weed/command/mount.go index 5ae83c481..6a28c2fe8 100644 --- a/weed/command/mount.go +++ b/weed/command/mount.go @@ -39,6 +39,8 @@ type MountOptions struct { debugFuse *bool localSocket *string disableXAttr *bool + windowsUid *int + windowsGid *int extraOptions []string fuseCommandPid int @@ -54,11 +56,11 @@ type MountOptions struct { rdmaTimeoutMs *int // Peer chunk sharing options (design-weed-mount-peer-chunk-sharing.md). - peerEnabled *bool - peerListen *string - peerAdvertise *string - peerDataCenter *string - peerRack *string + peerEnabled *bool + peerListen *string + peerAdvertise *string + peerDataCenter *string + peerRack *string dirIdleEvictSec *int @@ -127,6 +129,8 @@ func init() { mountOptions.debugFuse = cmdMount.Flag.Bool("debug.fuse", false, "log raw FUSE protocol requests and responses") mountOptions.localSocket = cmdMount.Flag.String("localSocket", "", "default to /tmp/seaweedfs-mount-.sock") mountOptions.disableXAttr = cmdMount.Flag.Bool("disableXAttr", false, "disable xattr") + mountOptions.windowsUid = cmdMount.Flag.Int("windows.uid", 0, "windows only: uid recorded on entries this mount creates, which other clients read") + mountOptions.windowsGid = cmdMount.Flag.Int("windows.gid", 0, "windows only: gid recorded on entries this mount creates, which other clients read") mountOptions.hasAutofs = cmdMount.Flag.Bool("autofs", false, "ignore autofs mounted on the same mountpoint (useful when systemd.automount and autofs is used)") mountOptions.fuseCommandPid = 0 diff --git a/weed/command/mount_common.go b/weed/command/mount_common.go new file mode 100644 index 000000000..0cd7450d5 --- /dev/null +++ b/weed/command/mount_common.go @@ -0,0 +1,277 @@ +//go:build linux || darwin || freebsd || windows + +package command + +import ( + "context" + "fmt" + "net" + "net/http" + "os" + "path" + "runtime" + "strconv" + "strings" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/reflection" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/mount" + "github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/mount_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/grace" +) + +func runMount(cmd *Command, args []string) bool { + + if *mountOptions.debug { + go http.ListenAndServe(fmt.Sprintf(":%d", *mountOptions.debugPort), nil) + } + + *mountCpuProfile = util.ResolvePath(*mountCpuProfile) + *mountMemProfile = util.ResolvePath(*mountMemProfile) + grace.SetupProfiling(*mountCpuProfile, *mountMemProfile) + if *mountReadRetryTime < time.Second { + *mountReadRetryTime = time.Second + } + util.RetryWaitTime = *mountReadRetryTime + + // 32 bits, not 64: os.FileMode is uint32, so a wider parse would let a + // nonsense umask truncate silently instead of being rejected here. + umask, umaskErr := strconv.ParseUint(*mountOptions.umaskString, 8, 32) + if umaskErr != nil { + fmt.Printf("can not parse umask %s", *mountOptions.umaskString) + return false + } + + if len(args) > 0 { + return false + } + + return RunMount(&mountOptions, os.FileMode(umask)) +} + +func ensureBucketAllowEmptyFolders(ctx context.Context, filerClient filer_pb.FilerClient, mountRoot, bucketRootPath string) error { + bucketPath, isBucketRootMount := bucketPathForMountRoot(mountRoot, bucketRootPath) + if !isBucketRootMount { + return nil + } + + entry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(bucketPath)) + if err != nil { + return err + } + if entry == nil { + return fmt.Errorf("bucket %s not found", bucketPath) + } + + if entry.Extended == nil { + entry.Extended = make(map[string][]byte) + } + if strings.EqualFold(strings.TrimSpace(string(entry.Extended[s3_constants.ExtAllowEmptyFolders])), "true") { + return nil + } + + entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("true") + + bucketFullPath := util.FullPath(bucketPath) + parent, _ := bucketFullPath.DirAndName() + if err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{ + Directory: parent, + Entry: entry, + }) + }); err != nil { + return err + } + + glog.V(3).Infof("RunMount: set bucket %s %s=true", bucketPath, s3_constants.ExtAllowEmptyFolders) + return nil +} + +func bucketPathForMountRoot(mountRoot, bucketRootPath string) (string, bool) { + cleanPath := path.Clean("/" + strings.TrimPrefix(mountRoot, "/")) + cleanBucketRoot := path.Clean("/" + strings.TrimPrefix(bucketRootPath, "/")) + if cleanBucketRoot == "/" { + return "", false + } + prefix := cleanBucketRoot + "/" + if !strings.HasPrefix(cleanPath, prefix) { + return "", false + } + rest := strings.TrimPrefix(cleanPath, prefix) + + bucketParts := strings.Split(rest, "/") + if len(bucketParts) != 1 || bucketParts[0] == "" { + return "", false + } + return cleanBucketRoot + "/" + bucketParts[0], true +} + +func peerStringOrEmpty(p *string) string { + if p == nil { + return "" + } + return *p +} + +// connectToFiler retries the filer handshake, returning the cluster's cipher +// setting and bucket root. +func connectToFiler(option *MountOptions) (filerAddresses []pb.ServerAddress, grpcDialOption grpc.DialOption, cipher bool, bucketRootPath string, ok bool) { + // try to connect to filer + filerAddresses = pb.ServerAddresses(*option.filer).ToAddresses() + util.LoadSecurityConfiguration() + grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client") + var err error + for i := 0; i < 10; i++ { + err = pb.WithOneOfGrpcFilerClients(false, filerAddresses, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { + resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}) + if err != nil { + return fmt.Errorf("get filer grpc address %v configuration: %w", filerAddresses, err) + } + cipher = resp.Cipher + bucketRootPath = resp.DirBuckets + return nil + }) + if err == nil { + break + } + glog.V(0).Infof("failed to talk to filer %v: %v", filerAddresses, err) + glog.V(0).Infof("wait for %d seconds ...", i+1) + time.Sleep(time.Duration(i+1) * time.Second) + } + if err != nil { + glog.Errorf("failed to talk to filer %v: %v", filerAddresses, err) + return nil, nil, false, "", false + } + if bucketRootPath == "" { + bucketRootPath = "/buckets" + } + return filerAddresses, grpcDialOption, cipher, bucketRootPath, true +} + +// fileSystemParams are the pieces of a mount that each platform works out for +// itself: where it is attached, and whose identity the entries carry. +type fileSystemParams struct { + dir string + mountRoot string + filerAddresses []pb.ServerAddress + grpcDialOption grpc.DialOption + cipher bool + uidGidMapper *meta_cache.UidGidMapper + uid uint32 + gid uint32 + mountMode os.FileMode + mountCtime time.Time + umask os.FileMode + chunkSizeLimitMB int + cacheDirForRead string + cacheDirForWrite string +} + +func buildSeaweedFileSystem(option *MountOptions, p fileSystemParams) *mount.WFS { + return mount.NewSeaweedFileSystem(&mount.Option{ + MountDirectory: p.dir, + FilerAddresses: p.filerAddresses, + GrpcDialOption: p.grpcDialOption, + FilerSigningKey: security.SigningKey(util.GetViper().GetString("jwt.filer_signing.key")), + FilerSigningExpiresAfterSec: util.GetViper().GetInt("jwt.filer_signing.expires_after_seconds"), + FilerMountRootPath: p.mountRoot, + Collection: *option.collection, + Replication: *option.replication, + TtlSec: int32(*option.ttlSec), + DiskType: types.ToDiskType(*option.diskType), + ChunkSizeLimit: int64(p.chunkSizeLimitMB) * 1024 * 1024, + ConcurrentWriters: *option.concurrentWriters, + ConcurrentReaders: *option.concurrentReaders, + CacheDirForRead: p.cacheDirForRead, + CacheSizeMBForRead: *option.cacheSizeMBForRead, + CacheDirForWrite: p.cacheDirForWrite, + WriteBufferSizeMB: *option.writeBufferSizeMB, + CacheMetaTTlSec: *option.cacheMetaTtlSec, + DataCenter: *option.dataCenter, + Quota: int64(*option.collectionQuota) * 1024 * 1024, + LogicalDiskUsage: *option.logicalDiskUsage, + MountUid: p.uid, + MountGid: p.gid, + MountMode: p.mountMode, + MountCtime: p.mountCtime, + MountMtime: time.Now(), + Umask: p.umask, + VolumeServerAccess: *mountOptions.volumeServerAccess, + Cipher: p.cipher, + UidGidMapper: p.uidGidMapper, + IncludeSystemEntries: *option.includeSystemEntries, + DefaultPermissions: *option.defaultPermissions, + DisableXAttr: *option.disableXAttr, + IsMacOs: runtime.GOOS == "darwin", + MetadataFlushSeconds: *option.metadataFlushSeconds, + // RDMA acceleration options + RdmaEnabled: *option.rdmaEnabled, + RdmaSidecarAddr: *option.rdmaSidecarAddr, + RdmaFallback: *option.rdmaFallback, + RdmaReadOnly: *option.rdmaReadOnly, + RdmaMaxConcurrent: *option.rdmaMaxConcurrent, + RdmaTimeoutMs: *option.rdmaTimeoutMs, + DirIdleEvictSec: *option.dirIdleEvictSec, + EnableDistributedLock: option.distributedLock != nil && *option.distributedLock, + WritebackCache: option.writebackCache != nil && *option.writebackCache, + PosixDirNlink: option.posixDirNlink != nil && *option.posixDirNlink, + // Peer chunk sharing + PeerEnabled: option.peerEnabled != nil && *option.peerEnabled, + PeerListen: peerStringOrEmpty(option.peerListen), + PeerAdvertise: peerStringOrEmpty(option.peerAdvertise), + PeerDataCenter: peerStringOrEmpty(option.peerDataCenter), + PeerRack: peerStringOrEmpty(option.peerRack), + }) +} + +// createMountRoot makes the filer-side directory the mount is rooted at. +func createMountRoot(wfs *mount.WFS, mountRoot, bucketRootPath string, filerAddresses []pb.ServerAddress) bool { + mountRootPath := util.FullPath(mountRoot) + mountRootParent, mountDir := mountRootPath.DirAndName() + if err := filer_pb.Mkdir(context.Background(), wfs, mountRootParent, mountDir, nil); err != nil { + fmt.Printf("failed to create dir %s on filer %s: %v\n", mountRoot, filerAddresses, err) + return false + } + if err := ensureBucketAllowEmptyFolders(context.Background(), wfs, mountRoot, bucketRootPath); err != nil { + fmt.Printf("failed to set bucket auto-remove-empty-folders policy for %s: %v\n", mountRoot, err) + return false + } + return true +} + +// serveMountGrpc exposes the local control socket used by "weed mount.stats". +func serveMountGrpc(wfs *mount.WFS, listener net.Listener) { + grpcS := pb.NewGrpcServer() + mount_pb.RegisterSeaweedMountServer(grpcS, wfs) + reflection.Register(grpcS) + go grpcS.Serve(listener) +} + +// resolveMountRoot trims the trailing slash the filer path must not carry. +func resolveMountRoot(filerMountRootPath string) string { + mountRoot := filerMountRootPath + if mountRoot != "/" && strings.HasSuffix(mountRoot, "/") { + mountRoot = mountRoot[0 : len(mountRoot)-1] + } + return mountRoot +} + +// resolveCacheDirs falls back to the read cache when no write cache is set. +func resolveCacheDirs(option *MountOptions) (string, string) { + cacheDirForRead := util.ResolvePath(*option.cacheDirForRead) + cacheDirForWrite := util.ResolvePath(*option.cacheDirForWrite) + if cacheDirForWrite == "" { + cacheDirForWrite = cacheDirForRead + } + return cacheDirForRead, cacheDirForWrite +} diff --git a/weed/command/mount_notsupported.go b/weed/command/mount_notsupported.go index da2a90cd8..1226b94d0 100644 --- a/weed/command/mount_notsupported.go +++ b/weed/command/mount_notsupported.go @@ -1,4 +1,4 @@ -//go:build !linux && !darwin && !freebsd +//go:build !linux && !darwin && !freebsd && !windows package command diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go index 3f357cedc..b352aea95 100644 --- a/weed/command/mount_std.go +++ b/weed/command/mount_std.go @@ -3,122 +3,26 @@ package command import ( - "context" "fmt" "net" - "net/http" "os" "os/user" - "path" "runtime" "strconv" "strings" "syscall" - "time" "github.com/seaweedfs/seaweedfs/weed/util/version" "github.com/seaweedfs/go-fuse/v2/fuse" "github.com/seaweedfs/seaweedfs/weed/glog" - "github.com/seaweedfs/seaweedfs/weed/mount" "github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" "github.com/seaweedfs/seaweedfs/weed/mount/unmount" - "github.com/seaweedfs/seaweedfs/weed/pb" - "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" - "github.com/seaweedfs/seaweedfs/weed/pb/mount_pb" - "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" - "github.com/seaweedfs/seaweedfs/weed/security" - "github.com/seaweedfs/seaweedfs/weed/storage/types" - "google.golang.org/grpc/reflection" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/util/grace" ) -func runMount(cmd *Command, args []string) bool { - - if *mountOptions.debug { - go http.ListenAndServe(fmt.Sprintf(":%d", *mountOptions.debugPort), nil) - } - - *mountCpuProfile = util.ResolvePath(*mountCpuProfile) - *mountMemProfile = util.ResolvePath(*mountMemProfile) - grace.SetupProfiling(*mountCpuProfile, *mountMemProfile) - if *mountReadRetryTime < time.Second { - *mountReadRetryTime = time.Second - } - util.RetryWaitTime = *mountReadRetryTime - - umask, umaskErr := strconv.ParseUint(*mountOptions.umaskString, 8, 64) - if umaskErr != nil { - fmt.Printf("can not parse umask %s", *mountOptions.umaskString) - return false - } - - if len(args) > 0 { - return false - } - - return RunMount(&mountOptions, os.FileMode(umask)) -} - -func ensureBucketAllowEmptyFolders(ctx context.Context, filerClient filer_pb.FilerClient, mountRoot, bucketRootPath string) error { - bucketPath, isBucketRootMount := bucketPathForMountRoot(mountRoot, bucketRootPath) - if !isBucketRootMount { - return nil - } - - entry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(bucketPath)) - if err != nil { - return err - } - if entry == nil { - return fmt.Errorf("bucket %s not found", bucketPath) - } - - if entry.Extended == nil { - entry.Extended = make(map[string][]byte) - } - if strings.EqualFold(strings.TrimSpace(string(entry.Extended[s3_constants.ExtAllowEmptyFolders])), "true") { - return nil - } - - entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("true") - - bucketFullPath := util.FullPath(bucketPath) - parent, _ := bucketFullPath.DirAndName() - if err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{ - Directory: parent, - Entry: entry, - }) - }); err != nil { - return err - } - - glog.V(3).Infof("RunMount: set bucket %s %s=true", bucketPath, s3_constants.ExtAllowEmptyFolders) - return nil -} - -func bucketPathForMountRoot(mountRoot, bucketRootPath string) (string, bool) { - cleanPath := path.Clean("/" + strings.TrimPrefix(mountRoot, "/")) - cleanBucketRoot := path.Clean("/" + strings.TrimPrefix(bucketRootPath, "/")) - if cleanBucketRoot == "/" { - return "", false - } - prefix := cleanBucketRoot + "/" - if !strings.HasPrefix(cleanPath, prefix) { - return "", false - } - rest := strings.TrimPrefix(cleanPath, prefix) - - bucketParts := strings.Split(rest, "/") - if len(bucketParts) != 1 || bucketParts[0] == "" { - return "", false - } - return cleanBucketRoot + "/" + bucketParts[0], true -} - func RunMount(option *MountOptions, umask os.FileMode) bool { // basic checks @@ -128,36 +32,10 @@ func RunMount(option *MountOptions, umask os.FileMode) bool { return false } - // try to connect to filer - filerAddresses := pb.ServerAddresses(*option.filer).ToAddresses() - util.LoadSecurityConfiguration() - grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client") - var cipher bool - var bucketRootPath string - var err error - for i := 0; i < 10; i++ { - err = pb.WithOneOfGrpcFilerClients(false, filerAddresses, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}) - if err != nil { - return fmt.Errorf("get filer grpc address %v configuration: %w", filerAddresses, err) - } - cipher = resp.Cipher - bucketRootPath = resp.DirBuckets - return nil - }) - if err != nil { - glog.V(0).Infof("failed to talk to filer %v: %v", filerAddresses, err) - glog.V(0).Infof("wait for %d seconds ...", i+1) - time.Sleep(time.Duration(i+1) * time.Second) - } - } - if err != nil { - glog.Errorf("failed to talk to filer %v: %v", filerAddresses, err) + filerAddresses, grpcDialOption, cipher, bucketRootPath, ok := connectToFiler(option) + if !ok { return true } - if bucketRootPath == "" { - bucketRootPath = "/buckets" - } filerMountRootPath := *option.filerMountRootPath @@ -316,82 +194,27 @@ func RunMount(option *MountOptions, umask os.FileMode) bool { fuseMountOptions.EnableSymlinkCaching = true } - // find mount point - mountRoot := filerMountRootPath - if mountRoot != "/" && strings.HasSuffix(mountRoot, "/") { - mountRoot = mountRoot[0 : len(mountRoot)-1] - } + mountRoot := resolveMountRoot(filerMountRootPath) + cacheDirForRead, cacheDirForWrite := resolveCacheDirs(option) - cacheDirForRead := util.ResolvePath(*option.cacheDirForRead) - cacheDirForWrite := util.ResolvePath(*option.cacheDirForWrite) - if cacheDirForWrite == "" { - cacheDirForWrite = cacheDirForRead - } - - seaweedFileSystem := mount.NewSeaweedFileSystem(&mount.Option{ - MountDirectory: dir, - FilerAddresses: filerAddresses, - GrpcDialOption: grpcDialOption, - FilerSigningKey: security.SigningKey(util.GetViper().GetString("jwt.filer_signing.key")), - FilerSigningExpiresAfterSec: util.GetViper().GetInt("jwt.filer_signing.expires_after_seconds"), - FilerMountRootPath: mountRoot, - Collection: *option.collection, - Replication: *option.replication, - TtlSec: int32(*option.ttlSec), - DiskType: types.ToDiskType(*option.diskType), - ChunkSizeLimit: int64(chunkSizeLimitMB) * 1024 * 1024, - ConcurrentWriters: *option.concurrentWriters, - ConcurrentReaders: *option.concurrentReaders, - CacheDirForRead: cacheDirForRead, - CacheSizeMBForRead: *option.cacheSizeMBForRead, - CacheDirForWrite: cacheDirForWrite, - WriteBufferSizeMB: *option.writeBufferSizeMB, - CacheMetaTTlSec: *option.cacheMetaTtlSec, - DataCenter: *option.dataCenter, - Quota: int64(*option.collectionQuota) * 1024 * 1024, - LogicalDiskUsage: *option.logicalDiskUsage, - MountUid: uid, - MountGid: gid, - MountMode: mountMode, - MountCtime: fileInfo.ModTime(), - MountMtime: time.Now(), - Umask: umask, - VolumeServerAccess: *mountOptions.volumeServerAccess, - Cipher: cipher, - UidGidMapper: uidGidMapper, - IncludeSystemEntries: *option.includeSystemEntries, - DefaultPermissions: *option.defaultPermissions, - DisableXAttr: *option.disableXAttr, - IsMacOs: runtime.GOOS == "darwin", - MetadataFlushSeconds: *option.metadataFlushSeconds, - // RDMA acceleration options - RdmaEnabled: *option.rdmaEnabled, - RdmaSidecarAddr: *option.rdmaSidecarAddr, - RdmaFallback: *option.rdmaFallback, - RdmaReadOnly: *option.rdmaReadOnly, - RdmaMaxConcurrent: *option.rdmaMaxConcurrent, - RdmaTimeoutMs: *option.rdmaTimeoutMs, - DirIdleEvictSec: *option.dirIdleEvictSec, - EnableDistributedLock: option.distributedLock != nil && *option.distributedLock, - WritebackCache: option.writebackCache != nil && *option.writebackCache, - PosixDirNlink: option.posixDirNlink != nil && *option.posixDirNlink, - // Peer chunk sharing - PeerEnabled: option.peerEnabled != nil && *option.peerEnabled, - PeerListen: peerStringOrEmpty(option.peerListen), - PeerAdvertise: peerStringOrEmpty(option.peerAdvertise), - PeerDataCenter: peerStringOrEmpty(option.peerDataCenter), - PeerRack: peerStringOrEmpty(option.peerRack), + seaweedFileSystem := buildSeaweedFileSystem(option, fileSystemParams{ + dir: dir, + mountRoot: mountRoot, + filerAddresses: filerAddresses, + grpcDialOption: grpcDialOption, + cipher: cipher, + uidGidMapper: uidGidMapper, + uid: uid, + gid: gid, + mountMode: mountMode, + mountCtime: fileInfo.ModTime(), + umask: umask, + chunkSizeLimitMB: chunkSizeLimitMB, + cacheDirForRead: cacheDirForRead, + cacheDirForWrite: cacheDirForWrite, }) - // create mount root - mountRootPath := util.FullPath(mountRoot) - mountRootParent, mountDir := mountRootPath.DirAndName() - if err = filer_pb.Mkdir(context.Background(), seaweedFileSystem, mountRootParent, mountDir, nil); err != nil { - fmt.Printf("failed to create dir %s on filer %s: %v\n", mountRoot, filerAddresses, err) - return false - } - if err := ensureBucketAllowEmptyFolders(context.Background(), seaweedFileSystem, mountRoot, bucketRootPath); err != nil { - fmt.Printf("failed to set bucket auto-remove-empty-folders policy for %s: %v\n", mountRoot, err) + if !createMountRoot(seaweedFileSystem, mountRoot, bucketRootPath, filerAddresses) { return false } @@ -416,10 +239,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool { } } - grpcS := pb.NewGrpcServer() - mount_pb.RegisterSeaweedMountServer(grpcS, seaweedFileSystem) - reflection.Register(grpcS) - go grpcS.Serve(montSocketListener) + serveMountGrpc(seaweedFileSystem, montSocketListener) err = seaweedFileSystem.StartBackgroundTasks() if err != nil { @@ -440,10 +260,3 @@ func RunMount(option *MountOptions, umask os.FileMode) bool { return true } - -func peerStringOrEmpty(p *string) string { - if p == nil { - return "" - } - return *p -} diff --git a/weed/command/mount_windows.go b/weed/command/mount_windows.go new file mode 100644 index 000000000..ba29e1374 --- /dev/null +++ b/weed/command/mount_windows.go @@ -0,0 +1,188 @@ +package command + +import ( + "fmt" + "net" + "os" + "path/filepath" + "runtime" + "strings" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" + "github.com/seaweedfs/seaweedfs/weed/mount/winfsp" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/grace" + "github.com/seaweedfs/seaweedfs/weed/util/version" +) + +// ownedByMounter reports the mount root as belonging to whoever started it, +// matching the uid=-1 option handed to WinFsp. It is a display value only: +// WinFsp substitutes the calling user, and it must never be persisted, since +// every other client would read the entry as owned by uid 4294967295. +const ownedByMounter = ^uint32(0) + +// windowsAttrTimeoutSec bounds how long WinFsp may serve cached attributes. +// Nothing invalidates that cache from this side, so it stays short rather than +// borrowing an unrelated flag's value. +const windowsAttrTimeoutSec = 1.0 + +func RunMount(option *MountOptions, umask os.FileMode) bool { + chunkSizeLimitMB := *mountOptions.chunkSizeLimitMB + if chunkSizeLimitMB <= 0 { + fmt.Printf("Please specify a reasonable buffer size.\n") + return false + } + + dir := *option.dir + if dir == "" { + fmt.Printf("Please specify the mount point via \"-dir\", for example -dir=S:\n") + return false + } + // A drive letter or a not-yet-existing directory is what WinFsp wants, so + // the mount point deliberately goes through none of the unix preparation: + // no ResolvePath, no auto-create, no stat. + if err := checkWindowsMountPoint(dir); err != nil { + fmt.Printf("%v\n", err) + return false + } + + filerAddresses, grpcDialOption, cipher, bucketRootPath, ok := connectToFiler(option) + if !ok { + return true + } + + if *option.localSocket == "" { + mountDirHash := util.HashToInt32([]byte(dir)) + if mountDirHash < 0 { + mountDirHash = -mountDirHash + } + *option.localSocket = filepath.Join(os.TempDir(), fmt.Sprintf("seaweedfs-mount-%d.sock", mountDirHash)) + } + if err := os.Remove(*option.localSocket); err != nil && !os.IsNotExist(err) { + glog.Fatalf("Failed to remove %s, error: %s", *option.localSocket, err.Error()) + } + mountSocketListener, err := net.Listen("unix", *option.localSocket) + if err != nil { + glog.Fatalf("Failed to listen on %s: %v", *option.localSocket, err) + } + + uidGidMapper, err := meta_cache.NewUidGidMapper(*option.uidMap, *option.gidMap) + if err != nil { + fmt.Printf("failed to parse %s %s: %v\n", *option.uidMap, *option.gidMap, err) + return false + } + + mountRoot := resolveMountRoot(*option.filerMountRootPath) + cacheDirForRead, cacheDirForWrite := resolveCacheDirs(option) + + seaweedFileSystem := buildSeaweedFileSystem(option, fileSystemParams{ + dir: dir, + mountRoot: mountRoot, + filerAddresses: filerAddresses, + grpcDialOption: grpcDialOption, + cipher: cipher, + uidGidMapper: uidGidMapper, + uid: uint32(*option.windowsUid), + gid: uint32(*option.windowsGid), + mountMode: os.ModeDir | 0777, + mountCtime: time.Now(), + umask: umask, + chunkSizeLimitMB: chunkSizeLimitMB, + cacheDirForRead: cacheDirForRead, + cacheDirForWrite: cacheDirForWrite, + }) + + if !createMountRoot(seaweedFileSystem, mountRoot, bucketRootPath, filerAddresses) { + return false + } + + host := winfsp.New(seaweedFileSystem, winfsp.Options{ + VolumeName: strings.ReplaceAll(*option.filer, ",", "+"), + Uid: ownedByMounter, + Gid: ownedByMounter, + AttrTimeout: windowsAttrTimeoutSec, + ReadOnly: *option.readOnly, + Debug: *option.debugFuse, + ExtraOptions: option.extraOptions, + }) + + grace.OnInterrupt(func() { + // The signal handler exits the process as soon as the hooks return, so + // anything still queued has to be flushed here rather than after Serve. + // WaitForAsyncFlush is idempotent, so the post-Serve call below is + // harmless if both run. + host.Unmount() + seaweedFileSystem.WaitForAsyncFlush() + }) + + serveMountGrpc(seaweedFileSystem, mountSocketListener) + + if err := seaweedFileSystem.StartBackgroundTasks(); err != nil { + fmt.Printf("failed to start background tasks: %v\n", err) + return false + } + + glog.V(0).Infof("mounting %s%s to %v", *option.filer, mountRoot, dir) + glog.V(0).Infof("This is SeaweedFS version %s %s %s", version.Version(), runtime.GOOS, runtime.GOARCH) + glog.V(0).Infof("Windows mount is beta: hard links are unavailable and byte-range locks are not shared across mounts") + + if err := host.Serve(windowsMountPoint(dir)); err != nil { + glog.Errorf("%v", err) + return false + } + + seaweedFileSystem.WaitForAsyncFlush() + seaweedFileSystem.ClearCacheDir() + + return true +} + +// checkWindowsMountPoint rejects the mount points WinFsp cannot take, which is +// worth doing up front because its own failure is a bare false. +func checkWindowsMountPoint(dir string) error { + if isDriveLetter(dir) { + if _, err := os.Stat(strings.TrimRight(dir, `\/`) + `\`); err == nil { + return fmt.Errorf("drive %s is already in use", dir) + } + return nil + } + if strings.HasPrefix(dir, `\\`) { + return nil + } + + // WinFsp creates the directory itself, with FILE_CREATE, and deletes it + // again when the filesystem goes away. An existing one — empty or not — + // fails with "mount point in use". + if _, err := os.Stat(dir); err == nil { + return fmt.Errorf("mount point %s already exists; WinFsp creates the directory itself, so give it a path that does not exist yet", dir) + } else if !os.IsNotExist(err) { + return err + } + if _, err := os.Stat(filepath.Dir(dir)); err != nil { + return fmt.Errorf("parent of mount point %s does not exist", dir) + } + return nil +} + +// windowsMountPoint drops a trailing separator from a drive letter. WinFsp +// recognises a drive only as exactly two characters; "S:\\" falls through to +// its directory handling and the mount fails. +func windowsMountPoint(dir string) string { + if isDriveLetter(dir) { + return strings.TrimRight(dir, `\/`) + } + return dir +} + +// isDriveLetter accepts both "S:" and "S:\\"; the trailing separator is how +// the drive is usually written, and windowsMountPoint normalises it. +func isDriveLetter(dir string) bool { + trimmed := strings.TrimRight(dir, `\/`) + if len(trimmed) != 2 || trimmed[1] != ':' { + return false + } + c := trimmed[0] + return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') +} diff --git a/weed/mount/posix_file_lock_test.go b/weed/mount/posix_file_lock_test.go index eebf78403..5ba3098a8 100644 --- a/weed/mount/posix_file_lock_test.go +++ b/weed/mount/posix_file_lock_test.go @@ -1,3 +1,5 @@ +//go:build !windows + package mount import ( diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 99fd5deec..0f8c28d06 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -185,7 +185,8 @@ type WFS struct { // asyncFlushWg tracks pending background flush work items for writebackCache mode. // Must be waited on before unmount cleanup to prevent data loss. - asyncFlushWg sync.WaitGroup + asyncFlushWg sync.WaitGroup + asyncFlushClose sync.Once // asyncFlushCh is a bounded work queue for background flush operations. // A fixed pool of worker goroutines processes items from this channel, @@ -627,6 +628,14 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, } } + // About to trust the filer, so first let any async flush of a just-closed + // handle land: it would otherwise answer with pre-close metadata, a + // truncate's old size or a write's old chunks. The cache paths above are + // already consistent and must not pay this wait. + if inode, found := wfs.inodeToPath.GetInode(fullpath); found { + wfs.waitForPendingAsyncFlush(inode) + } + // Directory not cached - fetch directly from filer without caching the entire directory. glog.V(4).Infof("lookupEntry fetching from filer %s", fullpath) var entry *filer_pb.Entry @@ -667,6 +676,28 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, glog.V(4).Infof("lookupEntry found deferred entry in local cache %s", fullpath) return localEntry, entryVersion{tsNs: localVersionTsNs}, fuse.OK } + // Creating many files at once can push the directory past + // the hot threshold and evict it, which drops the local + // placeholder a deferred create left behind. The handle + // still holding the unflushed entry is authoritative for + // it, so read it from there rather than reporting a file + // that plainly exists as missing. + if fh, fhFound := wfs.fhMap.FindFileHandle(inode); fhFound { + // Async upload workers append chunks under this lock; + // hold it for reading so FromPbEntry does not walk the + // chunk slice mid-reallocation. + fh.entryLock.RLock() + pbEntry := fh.GetEntry().GetEntry() + var localEntry *filer.Entry + if pbEntry != nil { + localEntry = filer.FromPbEntry(dir, pbEntry) + } + fh.entryLock.RUnlock() + if localEntry != nil { + glog.V(4).Infof("lookupEntry found deferred entry on its open handle %s", fullpath) + return localEntry, entryVersion{}, fuse.OK + } + } } } if inodeFound { diff --git a/weed/mount/weedfs_async_flush.go b/weed/mount/weedfs_async_flush.go index 4be34245c..c7c3e8361 100644 --- a/weed/mount/weedfs_async_flush.go +++ b/weed/mount/weedfs_async_flush.go @@ -133,10 +133,14 @@ func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFul // Called before unmount cleanup to ensure no data is lost. func (wfs *WFS) WaitForAsyncFlush() { wfs.asyncFlushWg.Wait() - if wfs.asyncFlushCh != nil { - close(wfs.asyncFlushCh) - } - if wfs.streamMutate != nil { - wfs.streamMutate.Close() - } + // Shutdown reaches here from both the interrupt hook and the path that + // resumes once serving stops, so closing has to survive a second caller. + wfs.asyncFlushClose.Do(func() { + if wfs.asyncFlushCh != nil { + close(wfs.asyncFlushCh) + } + if wfs.streamMutate != nil { + wfs.streamMutate.Close() + } + }) } diff --git a/weed/mount/weedfs_posix_lock_routed_test.go b/weed/mount/weedfs_posix_lock_routed_test.go index cb4d11a0d..064ea49ee 100644 --- a/weed/mount/weedfs_posix_lock_routed_test.go +++ b/weed/mount/weedfs_posix_lock_routed_test.go @@ -1,3 +1,5 @@ +//go:build !windows + package mount import ( diff --git a/weed/mount/winfsp/errno.go b/weed/mount/winfsp/errno.go new file mode 100644 index 000000000..36de24bd3 --- /dev/null +++ b/weed/mount/winfsp/errno.go @@ -0,0 +1,118 @@ +// Package winfsp mounts a SeaweedFS filesystem on Windows through WinFsp. +// +// WinFsp speaks a path-based FUSE dialect, while weed/mount implements the +// inode-based raw protocol the Linux kernel uses. This package translates +// between the two so both platforms run the same filesystem code. +package winfsp + +import ( + "syscall" + + "github.com/seaweedfs/go-fuse/v2/fuse" +) + +// cgofuse decodes what these operations return using MSVC/UCRT numbering, +// which parts company with Linux above 35: ENAMETOOLONG is 38 there and 36 is +// EDEADLK, so borrowing the Linux values silently reported the wrong error. +// errno_windows_test.go pins each of these to cgofuse's own constant. +const ( + ePERM = 1 + eNOENT = 2 + eINTR = 4 + eIO = 5 + eNXIO = 6 + eBADF = 9 + eAGAIN = 11 + eNOMEM = 12 + eACCES = 13 + eBUSY = 16 + eEXIST = 17 + eXDEV = 18 + eNODEV = 19 + eNOTDIR = 20 + eISDIR = 21 + eINVAL = 22 + eNFILE = 23 + eMFILE = 24 + eFBIG = 27 + eNOSPC = 28 + eSPIPE = 29 + eROFS = 30 + eMLINK = 31 + ePIPE = 32 + eRANGE = 34 + + eNAMETOOLONG = 38 + eNOSYS = 40 + eNOTEMPTY = 41 + eLOOP = 114 + eNODATA = 120 + eNOTSUP = 129 +) + +// statusErrno is keyed by the running platform's errno values, since that is +// what the raw filesystem returns, and yields the POSIX number for the wire. +// Built rather than declared: platforms alias errnos differently (freebsd has +// no ENODATA, linux makes ENOATTR the same value), so entries can collide. +// First one wins, so the general codes keep their meaning. +var statusErrno = buildStatusErrno() + +func buildStatusErrno() map[fuse.Status]int { + table := []struct { + status fuse.Status + errno int + }{ + {fuse.Status(syscall.EPERM), ePERM}, + {fuse.Status(syscall.ENOENT), eNOENT}, + {fuse.Status(syscall.EINTR), eINTR}, + {fuse.Status(syscall.EIO), eIO}, + {fuse.Status(syscall.ENXIO), eNXIO}, + {fuse.Status(syscall.EBADF), eBADF}, + {fuse.Status(syscall.EAGAIN), eAGAIN}, + {fuse.Status(syscall.ENOMEM), eNOMEM}, + {fuse.Status(syscall.EACCES), eACCES}, + {fuse.Status(syscall.EBUSY), eBUSY}, + {fuse.Status(syscall.EEXIST), eEXIST}, + {fuse.Status(syscall.EXDEV), eXDEV}, + {fuse.Status(syscall.ENODEV), eNODEV}, + {fuse.Status(syscall.ENOTDIR), eNOTDIR}, + {fuse.Status(syscall.EISDIR), eISDIR}, + {fuse.Status(syscall.EINVAL), eINVAL}, + {fuse.Status(syscall.ENFILE), eNFILE}, + {fuse.Status(syscall.EMFILE), eMFILE}, + {fuse.Status(syscall.EFBIG), eFBIG}, + {fuse.Status(syscall.ENOSPC), eNOSPC}, + {fuse.Status(syscall.ESPIPE), eSPIPE}, + {fuse.Status(syscall.EROFS), eROFS}, + {fuse.Status(syscall.EMLINK), eMLINK}, + {fuse.Status(syscall.EPIPE), ePIPE}, + {fuse.Status(syscall.ERANGE), eRANGE}, + {fuse.Status(syscall.ENAMETOOLONG), eNAMETOOLONG}, + {fuse.Status(syscall.ENOSYS), eNOSYS}, + {fuse.Status(syscall.ENOTEMPTY), eNOTEMPTY}, + {fuse.Status(syscall.ELOOP), eLOOP}, + {fuse.ENOTSUP, eNOTSUP}, + {fuse.ENODATA, eNODATA}, + {fuse.ENOATTR, eNODATA}, + } + m := make(map[fuse.Status]int, len(table)) + for _, entry := range table { + if _, exists := m[entry.status]; !exists { + m[entry.status] = entry.errno + } + } + return m +} + +// toErrno converts a raw filesystem status into the negative errno WinFsp +// wants. Unrecognised failures become -EIO rather than passing through a +// number that means something else on the other side. +func toErrno(status fuse.Status) int { + if status == fuse.OK { + return 0 + } + if n, ok := statusErrno[status]; ok { + return -n + } + return -eIO +} diff --git a/weed/mount/winfsp/errno_test.go b/weed/mount/winfsp/errno_test.go new file mode 100644 index 000000000..e2848137d --- /dev/null +++ b/weed/mount/winfsp/errno_test.go @@ -0,0 +1,49 @@ +package winfsp + +import ( + "syscall" + "testing" + + "github.com/seaweedfs/go-fuse/v2/fuse" +) + +// The numbers have to match what cgofuse decodes on the far side, regardless +// of how the host platform spells its own errno constants. That numbering is +// MSVC's, which parts company with Linux above 35. +func TestToErrnoUsesCgofuseNumbering(t *testing.T) { + cases := []struct { + status fuse.Status + want int + }{ + {fuse.OK, 0}, + {fuse.ENOENT, -2}, + {fuse.EIO, -5}, + {fuse.EACCES, -13}, + {fuse.EINVAL, -22}, + {fuse.ENOSYS, -40}, + {fuse.Status(syscall.EEXIST), -17}, + {fuse.Status(syscall.ENOSPC), -28}, + {fuse.Status(syscall.ENOTEMPTY), -41}, + } + for _, c := range cases { + if got := toErrno(c.status); got != c.want { + t.Errorf("toErrno(%v) = %d, want %d", c.status, got, c.want) + } + } +} + +// An unmapped failure must not pass its raw number through: on Windows those +// are APPLICATION_ERROR offsets that mean something else entirely. +func TestToErrnoUnknownBecomesEIO(t *testing.T) { + if got := toErrno(fuse.Status(1 << 20)); got != -eIO { + t.Errorf("toErrno(unknown) = %d, want %d", got, -eIO) + } +} + +func TestToErrnoNeverReturnsPositive(t *testing.T) { + for _, status := range []fuse.Status{fuse.ENOENT, fuse.EPERM, fuse.EBUSY, fuse.Status(4242)} { + if got := toErrno(status); got > 0 { + t.Errorf("toErrno(%v) = %d, want <= 0", status, got) + } + } +} diff --git a/weed/mount/winfsp/errno_windows_test.go b/weed/mount/winfsp/errno_windows_test.go new file mode 100644 index 000000000..500b2cf07 --- /dev/null +++ b/weed/mount/winfsp/errno_windows_test.go @@ -0,0 +1,78 @@ +package winfsp + +import ( + "syscall" + "testing" + + cgofuse "github.com/winfsp/cgofuse/fuse" +) + +// The errno values are spelled out so the table stays portable and testable on +// any runner. This pins them to what cgofuse actually decodes, so a divergence +// is a failing test rather than an operation reporting an unrelated error. +func TestErrnoValuesMatchCgofuse(t *testing.T) { + for _, c := range []struct { + name string + ours int + want int + }{ + {"EPERM", ePERM, cgofuse.EPERM}, + {"ENOENT", eNOENT, cgofuse.ENOENT}, + {"EINTR", eINTR, cgofuse.EINTR}, + {"EIO", eIO, cgofuse.EIO}, + {"ENXIO", eNXIO, cgofuse.ENXIO}, + {"EBADF", eBADF, cgofuse.EBADF}, + {"EAGAIN", eAGAIN, cgofuse.EAGAIN}, + {"ENOMEM", eNOMEM, cgofuse.ENOMEM}, + {"EACCES", eACCES, cgofuse.EACCES}, + {"EBUSY", eBUSY, cgofuse.EBUSY}, + {"EEXIST", eEXIST, cgofuse.EEXIST}, + {"EXDEV", eXDEV, cgofuse.EXDEV}, + {"ENODEV", eNODEV, cgofuse.ENODEV}, + {"ENOTDIR", eNOTDIR, cgofuse.ENOTDIR}, + {"EISDIR", eISDIR, cgofuse.EISDIR}, + {"EINVAL", eINVAL, cgofuse.EINVAL}, + {"ENFILE", eNFILE, cgofuse.ENFILE}, + {"EMFILE", eMFILE, cgofuse.EMFILE}, + {"EFBIG", eFBIG, cgofuse.EFBIG}, + {"ENOSPC", eNOSPC, cgofuse.ENOSPC}, + {"ESPIPE", eSPIPE, cgofuse.ESPIPE}, + {"EROFS", eROFS, cgofuse.EROFS}, + {"EMLINK", eMLINK, cgofuse.EMLINK}, + {"EPIPE", ePIPE, cgofuse.EPIPE}, + {"ERANGE", eRANGE, cgofuse.ERANGE}, + {"ENAMETOOLONG", eNAMETOOLONG, cgofuse.ENAMETOOLONG}, + {"ENOSYS", eNOSYS, cgofuse.ENOSYS}, + {"ENOTEMPTY", eNOTEMPTY, cgofuse.ENOTEMPTY}, + {"ELOOP", eLOOP, cgofuse.ELOOP}, + {"ENODATA", eNODATA, cgofuse.ENODATA}, + {"ENOTSUP", eNOTSUP, cgofuse.ENOTSUP}, + } { + if c.ours != c.want { + t.Errorf("%s = %d, cgofuse uses %d", c.name, c.ours, c.want) + } + } +} + +// translateOpenFlags relies on cgofuse using MSVC's numbering while the raw +// filesystem reads Go's. Pin that, because a swap between O_EXCL and O_TRUNC +// would turn "fail if it exists" into "truncate it". +func TestOpenFlagTranslation(t *testing.T) { + for _, c := range []struct { + name string + in int + want uint32 + }{ + {"O_EXCL", cgofuse.O_EXCL, uint32(syscall.O_EXCL)}, + {"O_TRUNC", cgofuse.O_TRUNC, uint32(syscall.O_TRUNC)}, + {"O_CREAT", cgofuse.O_CREAT, uint32(syscall.O_CREAT)}, + {"O_APPEND", cgofuse.O_APPEND, uint32(syscall.O_APPEND)}, + } { + if got := translateOpenFlags(c.in); got != c.want { + t.Errorf("translateOpenFlags(%s=%#x) = %#x, want %#x", c.name, c.in, got, c.want) + } + } + if got := translateOpenFlags(cgofuse.O_RDWR); got != uint32(syscall.O_RDWR) { + t.Errorf("access mode not preserved: %#x", got) + } +} diff --git a/weed/mount/winfsp/fs_windows.go b/weed/mount/winfsp/fs_windows.go new file mode 100644 index 000000000..cc9814aca --- /dev/null +++ b/weed/mount/winfsp/fs_windows.go @@ -0,0 +1,739 @@ +package winfsp + +import ( + "sync" + "syscall" + + cgofuse "github.com/winfsp/cgofuse/fuse" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/mount" +) + +const ( + rootInode = 1 + + // WinFsp passes this when an operation carries no open handle. + noHandle = ^uint64(0) + + // utimeOmit is the nanosecond marker asking for a timestamp to be left as + // it is; cgofuse hands it through rather than resolving it. + utimeOmit = (1 << 30) - 2 + + // windowsEpochCutoff is 1601-01-02 in unix seconds. Anything at or below + // it is Windows' own epoch leaking through rather than a real time. + windowsEpochCutoff = -11644387200 + + // How many entries to pull from one readdir round before handing them to + // WinFsp. Bounded so a directory with millions of children does not + // materialise in one slice. + readdirBatch = 4096 +) + +// never is a nil channel: receiving blocks forever, which is what the raw +// operations expect from a caller that cannot cancel. +var never chan struct{} + +// WinFS presents a mount.WFS through the path-based interface WinFsp speaks. +type WinFS struct { + cgofuse.FileSystemBase + + wfs *mount.WFS + uid uint32 + gid uint32 + readOnly bool + + // An open handle holds the lookup reference for its inode until Release, + // the way the kernel keeps one for an open file. WinFsp hands back only + // the handle, and the raw filesystem reuses one handle for repeated opens + // of the same inode, so the references are counted. + mu sync.Mutex + fileInodes map[uint64]*handleRef + dirInodes map[uint64]*handleRef +} + +func NewWinFS(wfs *mount.WFS, uid, gid uint32, readOnly bool) *WinFS { + return &WinFS{ + wfs: wfs, + uid: uid, + gid: gid, + readOnly: readOnly, + fileInodes: make(map[uint64]*handleRef), + dirInodes: make(map[uint64]*handleRef), + } +} + +// handleRef is the lookup references an open handle is holding, one per open +// that has not yet been released. +type handleRef struct { + inode uint64 + count int +} + +// retain parks one reference under a handle. +func (w *WinFS) retain(table map[uint64]*handleRef, handle, inode uint64) { + if handle == noHandle { + return + } + var stranded uint64 + w.mu.Lock() + switch existing, found := table[handle]; { + case found && existing.inode == inode: + existing.count++ + case found: + // The handle was reissued for a different inode; its old reference + // would otherwise never be returned. + stranded = existing.inode + table[handle] = &handleRef{inode: inode, count: 1} + default: + table[handle] = &handleRef{inode: inode, count: 1} + } + w.mu.Unlock() + w.forget(stranded) +} + +// releaseRetained hands back one reference, reporting the inode only once the +// last open of that handle is gone, so a repeated release cannot forget twice. +func (w *WinFS) releaseRetained(table map[uint64]*handleRef, handle uint64) uint64 { + w.mu.Lock() + defer w.mu.Unlock() + existing, found := table[handle] + if !found { + return 0 + } + existing.count-- + if existing.count > 0 { + return 0 + } + delete(table, handle) + return existing.inode +} + +// caller identifies who the raw filesystem should record as the owner of +// anything it creates. Windows has no uid to pass through, so entries carry +// the identity the mount was started with rather than root. +// denied reports whether a modification should be refused outright, which is +// how a read-only mount is enforced: WinFsp discards its own "ro" option. +func (w *WinFS) denied() bool { return w.readOnly } + +// inodeForHandle reports the inode an open handle is holding, or 0. WinFsp +// keeps the path it opened with and never updates it across a rename, so a +// handle is the more reliable of the two. +func (w *WinFS) inodeForHandle(table map[uint64]*handleRef, handle uint64) uint64 { + if handle == noHandle { + return 0 + } + w.mu.Lock() + defer w.mu.Unlock() + if existing, found := table[handle]; found { + return existing.inode + } + return 0 +} + +// ptr is for the operations that take the header by pointer. +func ptr(h fuse.InHeader) *fuse.InHeader { return &h } + +func (w *WinFS) caller(inode uint64) fuse.InHeader { + return fuse.InHeader{ + NodeId: inode, + Caller: fuse.Caller{Owner: fuse.Owner{Uid: w.uid, Gid: w.gid}}, + } +} + +// lookupRef collects the references a resolution took. Every operation that +// hands back an EntryOut grants the caller one, which the Linux kernel returns +// with FORGET. WinFsp has no FORGET, so anything the adapter looks up it has +// to release itself or the mount's inode table grows for the life of the +// process. +type lookupRef struct { + w *WinFS + inodes []uint64 +} + +// release returns every reference still held. +func (r *lookupRef) release() { + if r == nil { + return + } + for _, inode := range r.inodes { + r.w.forget(inode) + } + r.inodes = nil +} + +// keepLast hands the reference on the final component to the caller, which +// becomes responsible for releasing it. An open handle takes it this way and +// gives it back on Release. +func (r *lookupRef) keepLast() { + if r != nil && len(r.inodes) > 0 { + r.inodes = r.inodes[:len(r.inodes)-1] + } +} + +// forget returns one reference. The root is never looked up, so it never holds +// one to give back. +func (w *WinFS) forget(inode uint64) { + if inode == rootInode || inode == 0 { + return + } + w.wfs.Forget(inode, 1) +} + +// walk resolves a path one component at a time. Lookup does more than find an +// inode: it refreshes what the mount knows about the entry, so the result of a +// preceding truncate or write is visible to the caller. +func (w *WinFS) walk(parts []string) (uint64, *lookupRef, fuse.Status) { + ref := &lookupRef{w: w} + inode := uint64(rootInode) + for _, name := range parts { + var out fuse.EntryOut + if status := w.wfs.Lookup(never, ptr(w.caller(inode)), name, &out); status != fuse.OK { + ref.release() + return 0, nil, status + } + inode = out.NodeId + ref.inodes = append(ref.inodes, inode) + } + return inode, ref, fuse.OK +} + +// resolve walks a WinFsp path down to an inode. The caller must release the +// returned reference. +func (w *WinFS) resolve(path string) (uint64, *lookupRef, fuse.Status) { + return w.walk(splitPath(path)) +} + +// resolveParent resolves everything but the last component, which the create +// and delete operations need separately. The caller must release the reference. +func (w *WinFS) resolveParent(path string) (uint64, string, *lookupRef, fuse.Status) { + parentParts, name, ok := splitParent(path) + if !ok { + return 0, "", nil, fuse.EINVAL + } + parent, ref, status := w.walk(parentParts) + if status != fuse.OK { + return 0, "", nil, status + } + return parent, name, ref, fuse.OK +} + +func (w *WinFS) attrToStat(attr *fuse.Attr, stat *cgofuse.Stat_t) { + stat.Ino = attr.Ino + stat.Mode = attr.Mode + stat.Nlink = attr.Nlink + stat.Size = int64(attr.Size) + stat.Blocks = int64(attr.Blocks) + stat.Blksize = int64(attr.Blksize) + stat.Uid = w.uid + stat.Gid = w.gid + stat.Atim = cgofuse.Timespec{Sec: int64(attr.Atime), Nsec: int64(attr.Atimensec)} + stat.Mtim = cgofuse.Timespec{Sec: int64(attr.Mtime), Nsec: int64(attr.Mtimensec)} + stat.Ctim = cgofuse.Timespec{Sec: int64(attr.Ctime), Nsec: int64(attr.Ctimensec)} + // Windows shows a creation time and has nothing to derive it from; ctime + // is the closest the filer tracks. + stat.Birthtim = stat.Ctim +} + +// translateOpenFlags converts cgofuse's open flags, which follow MSVC's +// numbering, into the values the raw filesystem tests against. Only the access +// mode and O_TRUNC happen to agree; O_EXCL would otherwise read as O_APPEND. +func translateOpenFlags(flags int) uint32 { + out := uint32(flags & cgofuse.O_ACCMODE) + for _, pair := range []struct { + from int + to int + }{ + {cgofuse.O_APPEND, syscall.O_APPEND}, + {cgofuse.O_CREAT, syscall.O_CREAT}, + {cgofuse.O_TRUNC, syscall.O_TRUNC}, + {cgofuse.O_EXCL, syscall.O_EXCL}, + } { + if flags&pair.from != 0 { + out |= uint32(pair.to) + } + } + return out +} + +// logResolveFailure keeps a missing file quiet. Windows probes for entries +// that do not exist as a matter of course, so ENOENT is an answer rather than +// a fault; anything else is worth a line. +func logResolveFailure(op, path string, status fuse.Status) { + if status == fuse.ENOENT { + glog.V(4).Infof("%s %s: no such entry", op, path) + return + } + glog.Errorf("%s %s: resolving: %v", op, path, status) +} + +func (w *WinFS) Statfs(path string, stat *cgofuse.Statfs_t) int { + var out fuse.StatfsOut + if status := w.wfs.StatFs(never, ptr(w.caller(rootInode)), &out); status != fuse.OK { + return toErrno(status) + } + stat.Bsize = uint64(out.Bsize) + stat.Frsize = uint64(out.Frsize) + stat.Blocks = out.Blocks + stat.Bfree = out.Bfree + stat.Bavail = out.Bavail + stat.Files = out.Files + stat.Ffree = out.Ffree + stat.Favail = out.Ffree + stat.Namemax = uint64(out.NameLen) + return 0 +} + +func (w *WinFS) Getattr(path string, stat *cgofuse.Stat_t, fh uint64) int { + inode := w.inodeForHandle(w.fileInodes, fh) + if inode == 0 { + var ref *lookupRef + var status fuse.Status + inode, ref, status = w.resolve(path) + if status != fuse.OK { + logResolveFailure("getattr", path, status) + return toErrno(status) + } + defer ref.release() + } + in := &fuse.GetAttrIn{InHeader: w.caller(inode)} + if fh != noHandle { + in.Fh_ = fh + in.Flags_ = fuse.FUSE_GETATTR_FH + } + var out fuse.AttrOut + if status := w.wfs.GetAttr(never, in, &out); status != fuse.OK { + return toErrno(status) + } + w.attrToStat(&out.Attr, stat) + return 0 +} + +func (w *WinFS) Mkdir(path string, mode uint32) int { + if w.denied() { + return -eROFS + } + parent, name, ref, status := w.resolveParent(path) + if status != fuse.OK { + return toErrno(status) + } + defer ref.release() + in := &fuse.MkdirIn{InHeader: w.caller(parent), Mode: mode} + var out fuse.EntryOut + status = w.wfs.Mkdir(never, in, name, &out) + if status == fuse.OK { + // Mkdir hands back an EntryOut, and nothing on this side will forget it. + w.forget(out.NodeId) + } + return toErrno(status) +} + +func (w *WinFS) Rmdir(path string) int { + if w.denied() { + return -eROFS + } + parent, name, ref, status := w.resolveParent(path) + if status != fuse.OK { + return toErrno(status) + } + defer ref.release() + return toErrno(w.wfs.Rmdir(never, ptr(w.caller(parent)), name)) +} + +func (w *WinFS) Unlink(path string) int { + if w.denied() { + return -eROFS + } + parent, name, ref, status := w.resolveParent(path) + if status != fuse.OK { + return toErrno(status) + } + defer ref.release() + return toErrno(w.wfs.Unlink(never, ptr(w.caller(parent)), name)) +} + +func (w *WinFS) Rename(oldpath string, newpath string) int { + if w.denied() { + return -eROFS + } + oldParent, oldName, oldRef, status := w.resolveParent(oldpath) + if status != fuse.OK { + return toErrno(status) + } + defer oldRef.release() + newParent, newName, newRef, status := w.resolveParent(newpath) + if status != fuse.OK { + return toErrno(status) + } + defer newRef.release() + in := &fuse.RenameIn{InHeader: w.caller(oldParent), Newdir: newParent} + return toErrno(w.wfs.Rename(never, in, oldName, newName)) +} + +func (w *WinFS) Create(path string, flags int, mode uint32) (int, uint64) { + if w.denied() { + return -eROFS, noHandle + } + parent, name, ref, status := w.resolveParent(path) + if status != fuse.OK { + glog.Errorf("create %s: resolving the parent directory: %v", path, status) + return toErrno(status), noHandle + } + defer ref.release() + in := &fuse.CreateIn{InHeader: w.caller(parent), Flags: translateOpenFlags(flags), Mode: mode} + var out fuse.CreateOut + if status := w.wfs.Create(never, in, name, &out); status != fuse.OK { + glog.Errorf("create %s in inode %d: %v", name, parent, status) + return toErrno(status), noHandle + } + // Create grants a reference on the new inode; hold it for as long as the + // handle lives and give it back in Release. + w.retain(w.fileInodes, out.Fh, out.NodeId) + return 0, out.Fh +} + +func (w *WinFS) Open(path string, flags int) (int, uint64) { + inode, ref, status := w.resolve(path) + if status != fuse.OK { + logResolveFailure("open", path, status) + return toErrno(status), noHandle + } + defer ref.release() + in := &fuse.OpenIn{InHeader: w.caller(inode), Flags: translateOpenFlags(flags)} + var out fuse.OpenOut + if status := w.wfs.Open(never, in, &out); status != fuse.OK { + glog.Errorf("open %s inode %d: %v", path, inode, status) + return toErrno(status), noHandle + } + ref.keepLast() + w.retain(w.fileInodes, out.Fh, inode) + return 0, out.Fh +} + +func (w *WinFS) Read(path string, buff []byte, ofst int64, fh uint64) int { + if fh == noHandle { + return -eBADF + } + in := &fuse.ReadIn{ + Fh: fh, + Offset: uint64(ofst), + Size: uint32(len(buff)), + } + result, status := w.wfs.Read(never, in, buff) + if status != fuse.OK { + return toErrno(status) + } + data, status := result.Bytes(buff) + if status != fuse.OK { + return toErrno(status) + } + if len(data) > 0 && &data[0] != &buff[0] { + copy(buff, data) + } + return len(data) +} + +func (w *WinFS) Write(path string, buff []byte, ofst int64, fh uint64) int { + if w.denied() { + return -eROFS + } + if fh == noHandle { + return -eBADF + } + in := &fuse.WriteIn{ + Fh: fh, + Offset: uint64(ofst), + Size: uint32(len(buff)), + } + written, status := w.wfs.Write(never, in, buff) + if status != fuse.OK { + if status == fuse.ENOENT { + glog.Errorf("write %s: handle %d is not open", path, fh) + } + return toErrno(status) + } + return int(written) +} + +func (w *WinFS) Truncate(path string, size int64, fh uint64) int { + if w.denied() { + return -eROFS + } + inode := w.inodeForHandle(w.fileInodes, fh) + if inode == 0 { + var ref *lookupRef + var status fuse.Status + inode, ref, status = w.resolve(path) + if status != fuse.OK { + return toErrno(status) + } + defer ref.release() + } + in := &fuse.SetAttrIn{} + in.NodeId = inode + in.Valid = fuse.FATTR_SIZE + in.Size = uint64(size) + if fh != noHandle { + in.Valid |= fuse.FATTR_FH + in.Fh = fh + } + var out fuse.AttrOut + return toErrno(w.wfs.SetAttr(never, in, &out)) +} + +func (w *WinFS) Chmod(path string, mode uint32) int { + if w.denied() { + return -eROFS + } + inode, ref, status := w.resolve(path) + if status != fuse.OK { + return toErrno(status) + } + defer ref.release() + in := &fuse.SetAttrIn{} + in.NodeId = inode + in.Valid = fuse.FATTR_MODE + in.Mode = mode + var out fuse.AttrOut + return toErrno(w.wfs.SetAttr(never, in, &out)) +} + +func (w *WinFS) Utimens(path string, tmsp []cgofuse.Timespec) int { + if w.denied() { + return -eROFS + } + inode, ref, status := w.resolve(path) + if status != fuse.OK { + return toErrno(status) + } + defer ref.release() + in := &fuse.SetAttrIn{} + in.NodeId = inode + if len(tmsp) < 2 { + in.Valid = fuse.FATTR_ATIME_NOW | fuse.FATTR_MTIME_NOW + } else { + // Windows sends times around its own 1601 epoch, which arrive here as + // a large negative second count and would be stored as a year-1601 + // timestamp that every other client then reads. Leave those alone. + if applyTimespec(tmsp[0]) { + in.Valid |= fuse.FATTR_ATIME + in.Atime, in.Atimensec = uint64(tmsp[0].Sec), uint32(tmsp[0].Nsec) + } + if applyTimespec(tmsp[1]) { + in.Valid |= fuse.FATTR_MTIME + in.Mtime, in.Mtimensec = uint64(tmsp[1].Sec), uint32(tmsp[1].Nsec) + } + if in.Valid == 0 { + return 0 + } + } + var out fuse.AttrOut + return toErrno(w.wfs.SetAttr(never, in, &out)) +} + +// applyTimespec reports whether a timestamp should be written. UTIME_OMIT asks +// for the existing value to be kept, and a time at or below Windows' own 1601 +// epoch arrives as a large negative second count that would be stored verbatim. +func applyTimespec(ts cgofuse.Timespec) bool { + return ts.Nsec != utimeOmit && ts.Sec > windowsEpochCutoff +} + +func (w *WinFS) Flush(path string, fh uint64) int { + if fh == noHandle { + return 0 + } + return toErrno(w.wfs.Flush(never, &fuse.FlushIn{InHeader: w.caller(0), Fh: fh})) +} + +func (w *WinFS) Fsync(path string, datasync bool, fh uint64) int { + if fh == noHandle { + return 0 + } + return toErrno(w.wfs.Fsync(never, &fuse.FsyncIn{Fh: fh})) +} + +func (w *WinFS) Release(path string, fh uint64) int { + if fh == noHandle { + return 0 + } + w.wfs.Release(never, &fuse.ReleaseIn{Fh: fh}) + w.forget(w.releaseRetained(w.fileInodes, fh)) + return 0 +} + +func (w *WinFS) Opendir(path string) (int, uint64) { + inode, ref, status := w.resolve(path) + if status != fuse.OK { + return toErrno(status), noHandle + } + defer ref.release() + var out fuse.OpenOut + if status := w.wfs.OpenDir(never, &fuse.OpenIn{InHeader: w.caller(inode)}, &out); status != fuse.OK { + return toErrno(status), noHandle + } + ref.keepLast() + w.retain(w.dirInodes, out.Fh, inode) + return 0, out.Fh +} + +func (w *WinFS) Releasedir(path string, fh uint64) int { + if fh == noHandle { + return 0 + } + w.wfs.ReleaseDir(&fuse.ReleaseIn{Fh: fh}) + w.forget(w.releaseRetained(w.dirInodes, fh)) + return 0 +} + +// readdirSink collects one readdir round. The raw operation fills the returned +// EntryOut after AddEntryPlus returns, so nothing can be converted until the +// round is over. +type readdirSink struct { + names []string + offsets []uint64 + inodes []uint64 + attrs []*fuse.EntryOut + limit int + + // lastOffset advances over dropped entries too, so a batch that is all + // dot entries still moves the enumeration along. + lastOffset uint64 + seen int + + // discard absorbs the attributes of an entry that is being dropped; the + // raw filesystem fills the block after handing it back. + discard fuse.EntryOut +} + +// The kernel expects readdir to report "." and "..", but Windows enumerates a +// directory without them and shows whatever it is given, so they are dropped +// rather than surfaced as two extra children. +func isDotEntry(name string) bool { + return name == "." || name == ".." +} + +func (s *readdirSink) AddEntry(entry fuse.DirEntry) bool { + if len(s.names) >= s.limit { + return false + } + s.seen++ + s.lastOffset = entry.Off + if isDotEntry(entry.Name) { + return true + } + s.names = append(s.names, entry.Name) + s.offsets = append(s.offsets, entry.Off) + s.inodes = append(s.inodes, entry.Ino) + s.attrs = append(s.attrs, nil) + return true +} + +func (s *readdirSink) AddEntryPlus(entry fuse.DirEntry) *fuse.EntryOut { + if len(s.names) >= s.limit { + return nil + } + s.seen++ + s.lastOffset = entry.Off + if isDotEntry(entry.Name) { + return &s.discard + } + out := &fuse.EntryOut{} + s.names = append(s.names, entry.Name) + s.offsets = append(s.offsets, entry.Off) + s.inodes = append(s.inodes, entry.Ino) + s.attrs = append(s.attrs, out) + return out +} + +func (w *WinFS) Readdir(path string, fill func(name string, stat *cgofuse.Stat_t, ofst int64) bool, ofst int64, fh uint64) int { + inode := w.inodeForHandle(w.dirInodes, fh) + if inode == 0 { + var ref *lookupRef + var status fuse.Status + inode, ref, status = w.resolve(path) + if status != fuse.OK { + return toErrno(status) + } + defer ref.release() + } + offset := uint64(ofst) + for { + sink := &readdirSink{limit: readdirBatch} + in := &fuse.ReadIn{ + InHeader: w.caller(inode), + Fh: fh, + Offset: offset, + Size: 1 << 20, + } + if status := w.wfs.ReadDirectoryInto(in, sink, true); status != fuse.OK { + return toErrno(status) + } + if sink.seen == 0 { + return 0 + } + filled := true + for i, name := range sink.names { + var stat cgofuse.Stat_t + var statp *cgofuse.Stat_t + if attr := sink.attrs[i]; attr != nil { + w.attrToStat(&attr.Attr, &stat) + statp = &stat + } + if filled && !fill(name, statp, int64(sink.offsets[i])) { + filled = false + } + } + // A readdirplus entry carries a reference of its own. Give every one + // of them back, including any the fill above stopped short of, or a + // single walk of a wide directory strands one reference per child. + for _, child := range sink.inodes { + w.forget(child) + } + if !filled { + return 0 + } + if sink.lastOffset <= offset { + return 0 + } + offset = sink.lastOffset + } +} + +func (w *WinFS) Readlink(path string) (int, string) { + // WinFsp probes the root to decide whether the volume has symlinks, and + // turns them on unless this fails. Leaving them on costs a getattr per + // path component on every open, for a feature Symlink already refuses. + if len(splitPath(path)) == 0 { + return -eNOSYS, "" + } + inode, ref, status := w.resolve(path) + if status != fuse.OK { + return toErrno(status), "" + } + defer ref.release() + target, status := w.wfs.Readlink(never, ptr(w.caller(inode))) + if status != fuse.OK { + return toErrno(status), "" + } + return 0, string(target) +} + +// Symlink is refused for now. The entry is easy to create, but WinFsp only +// follows it once the reparse point is wired up, so it would otherwise read +// back as an empty file. +func (w *WinFS) Symlink(target string, newpath string) int { + return -eNOSYS +} + +// Chown accepts and discards. Windows has no uid to record, but WinFsp passes +// a chown failure straight out of SetSecurity, so refusing it breaks Explorer's +// Security tab and icacls for changes that are not about ownership at all. +func (w *WinFS) Chown(path string, uid uint32, gid uint32) int { + return 0 +} + +// Link is not implemented: WinFsp has no hard links. +func (w *WinFS) Link(oldpath string, newpath string) int { + return -eNOSYS +} diff --git a/weed/mount/winfsp/host_windows.go b/weed/mount/winfsp/host_windows.go new file mode 100644 index 000000000..2c9c51638 --- /dev/null +++ b/weed/mount/winfsp/host_windows.go @@ -0,0 +1,109 @@ +package winfsp + +import ( + "fmt" + "strconv" + "strings" + + cgofuse "github.com/winfsp/cgofuse/fuse" + + "github.com/seaweedfs/seaweedfs/weed/mount" +) + +// Options are the WinFsp-specific knobs the mount command passes through. +type Options struct { + // VolumeName labels the drive in Explorer. + VolumeName string + + // Uid and Gid are reported for every entry. WinFsp overrides what is + // reported anyway (see the uid=-1 option below), so these are what gets + // written to the filer and read by every other client. + Uid uint32 + Gid uint32 + + // AttrTimeout bounds how long WinFsp caches attributes, in seconds. + // Nothing invalidates its cache, so this is the only coherence knob. + AttrTimeout float64 + + // ReadOnly rejects every modification. WinFsp has no "ro" option — it + // discards the flag and leaves the volume writable — so the refusal has + // to happen in the operations themselves. + ReadOnly bool + + // Debug turns on cgofuse's operation trace. + Debug bool + + // ExtraOptions are passed through to WinFsp as -o arguments. + ExtraOptions []string +} + +// Host is a WinFsp mount that has not been started yet. +type Host struct { + host *cgofuse.FileSystemHost + options Options +} + +// New wires wfs up to WinFsp. Nothing is mounted until Serve. +func New(wfs *mount.WFS, options Options) *Host { + host := cgofuse.NewFileSystemHost(NewWinFS(wfs, options.Uid, options.Gid, options.ReadOnly)) + host.SetCapReaddirPlus(true) + host.SetUseIno(true) + return &Host{host: host, options: options} +} + +// Serve attaches the filesystem at mountPoint, which is a drive letter ("S:"), +// a directory that does not yet exist, or a UNC path. It blocks until the +// filesystem is unmounted. +func (h *Host) Serve(mountPoint string) error { + opts := []string{ + "-o", "volname=" + h.volumeName(), + "-o", "uid=-1", + "-o", "gid=-1", + } + if h.options.AttrTimeout > 0 { + timeout := strconv.FormatFloat(h.options.AttrTimeout, 'f', -1, 64) + opts = append(opts, "-o", "attr_timeout="+timeout, "-o", "entry_timeout="+timeout) + } + if h.options.Debug { + opts = append(opts, "-d") + } + for _, extra := range h.options.ExtraOptions { + opts = append(opts, "-o", extra) + } + + if err := h.mount(mountPoint, opts); err != nil { + return err + } + return nil +} + +// mount turns a refusal into an error. cgofuse panics rather than returning +// when winfsp-x64.dll is missing, which is the most likely reason for a +// failure here and the one worth naming. +func (h *Host) mount(mountPoint string, opts []string) (err error) { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("mounting %s failed (%v); is WinFsp installed?", mountPoint, r) + } + }() + if !h.host.Mount(mountPoint, opts) { + return fmt.Errorf("WinFsp refused to mount %s; check that WinFsp is installed and the mount point is free", mountPoint) + } + return nil +} + +// Unmount detaches the filesystem, releasing a blocked Serve. +func (h *Host) Unmount() bool { + return h.host.Unmount() +} + +// volumeName keeps the label parseable: WinFsp splits options on commas, so a +// label carrying one would be cut short and take the rest of the option string +// with it. +func (h *Host) volumeName() string { + name := strings.ReplaceAll(h.options.VolumeName, ",", "+") + if name == "" { + return "SeaweedFS" + } + return name +} diff --git a/weed/mount/winfsp/path.go b/weed/mount/winfsp/path.go new file mode 100644 index 000000000..3250546f6 --- /dev/null +++ b/weed/mount/winfsp/path.go @@ -0,0 +1,33 @@ +package winfsp + +import "strings" + +// splitPath breaks a WinFsp path into the components to walk. WinFsp normally +// hands out forward slashes, but callers reach the mount with either separator, +// and empty or "." components have to drop out rather than become a lookup for +// a name that does not exist. +func splitPath(path string) []string { + trimmed := strings.Trim(strings.ReplaceAll(path, `\`, "/"), "/") + if trimmed == "" { + return nil + } + parts := make([]string, 0, strings.Count(trimmed, "/")+1) + for _, name := range strings.Split(trimmed, "/") { + if name == "" || name == "." { + continue + } + parts = append(parts, name) + } + return parts +} + +// splitParent separates the components leading up to the final name, which the +// create and delete operations need apart. ok is false for the root, which has +// no parent to operate in. +func splitParent(path string) (parent []string, name string, ok bool) { + parts := splitPath(path) + if len(parts) == 0 { + return nil, "", false + } + return parts[:len(parts)-1], parts[len(parts)-1], true +} diff --git a/weed/mount/winfsp/path_test.go b/weed/mount/winfsp/path_test.go new file mode 100644 index 000000000..8888ad1c5 --- /dev/null +++ b/weed/mount/winfsp/path_test.go @@ -0,0 +1,78 @@ +package winfsp + +import ( + "strings" + "testing" +) + +func TestSplitPath(t *testing.T) { + cases := []struct { + in string + want []string + }{ + {"/", nil}, + {"", nil}, + {"//", nil}, + {".", nil}, + {"/foo", []string{"foo"}}, + {"foo", []string{"foo"}}, + {"/foo/", []string{"foo"}}, + {"/foo/bar", []string{"foo", "bar"}}, + {"/foo//bar", []string{"foo", "bar"}}, + {"/foo/./bar", []string{"foo", "bar"}}, + {`\foo\bar`, []string{"foo", "bar"}}, + {`/foo\bar`, []string{"foo", "bar"}}, + {"/foo bar/baz qux", []string{"foo bar", "baz qux"}}, + {"/eñe/日本", []string{"eñe", "日本"}}, + // ".." is a name like any other here; the filer resolves it, and + // treating it structurally would let a path escape the mount root. + {"/foo/../bar", []string{"foo", "..", "bar"}}, + } + for _, c := range cases { + got := splitPath(c.in) + if len(got) != len(c.want) { + t.Errorf("splitPath(%q) = %v, want %v", c.in, got, c.want) + continue + } + for i := range got { + if got[i] != c.want[i] { + t.Errorf("splitPath(%q) = %v, want %v", c.in, got, c.want) + break + } + } + } +} + +func TestSplitParent(t *testing.T) { + cases := []struct { + in string + wantParent string + wantName string + wantOK bool + }{ + {"/", "", "", false}, + {"", "", "", false}, + {"/foo", "", "foo", true}, + {"/foo/bar", "foo", "bar", true}, + {"/foo/bar/baz", "foo/bar", "baz", true}, + {`\foo\bar`, "foo", "bar", true}, + {"/foo/bar/", "foo", "bar", true}, + } + for _, c := range cases { + parent, name, ok := splitParent(c.in) + if ok != c.wantOK || name != c.wantName || strings.Join(parent, "/") != c.wantParent { + t.Errorf("splitParent(%q) = (%q, %q, %v), want (%q, %q, %v)", + c.in, strings.Join(parent, "/"), name, ok, c.wantParent, c.wantName, c.wantOK) + } + } +} + +// The root has no parent, so operations that need one must be rejected rather +// than silently acting on the root itself. +func TestSplitParentRejectsRoot(t *testing.T) { + for _, root := range []string{"/", "", "//", `\`} { + if _, _, ok := splitParent(root); ok { + t.Errorf("splitParent(%q) accepted the root", root) + } + } +}