diff --git a/.kres.yaml b/.kres.yaml index 681017ce..cfc8bab9 100644 --- a/.kres.yaml +++ b/.kres.yaml @@ -45,6 +45,7 @@ spec: - mdadm - mei - mellanox-mstflint + - mergerfs - metal-agent - multipath-tools - nebula diff --git a/MAINTAINERS.md b/MAINTAINERS.md index 1f19ba09..6a55b15a 100644 --- a/MAINTAINERS.md +++ b/MAINTAINERS.md @@ -42,6 +42,7 @@ If the field is marked as `Needs Maintainer`, it means that the package is curre | lldpd | Nokia (Niklas Wik) | [salkin](https://github.com/salkin) | | mdadm | Serge Logvinov | [sergelogvinov](https://github.com/sergelogvinov) | | mei | Nick Meyer | [e3b0c442](https://github.com/e3b0c442) | +| mergerfs | Micke Lisinge | [micke](https://github.com/micke) | | metal-agent | Sidero Labs | NA | | multipath-tools | INS | [Untersander](https://github.com/Untersander), [SimLi1333](https://github.com/SimLi1333) | | nebula | s e | [iamwacko](https://github.com/iamwacko) | diff --git a/Makefile b/Makefile index 47982e3b..46ab0e82 100644 --- a/Makefile +++ b/Makefile @@ -109,6 +109,7 @@ TARGETS += lldpd TARGETS += mdadm TARGETS += mei TARGETS += mellanox-mstflint +TARGETS += mergerfs TARGETS += metal-agent TARGETS += multipath-tools TARGETS += nebula diff --git a/README.md b/README.md index 74fbe3c1..8cb0c0bd 100644 --- a/README.md +++ b/README.md @@ -165,6 +165,7 @@ tiers based on support level: | [fuse3](storage/fuse3) | :green_square: core | [ghcr.io/siderolabs/fuse3](https://github.com/siderolabs/extensions/pkgs/container/fuse3) | `3.18.3` | This system extension provides fuse3 functionality. | | [iscsi-tools](storage/iscsi-tools) | :green_square: core | [ghcr.io/siderolabs/iscsi-tools](https://github.com/siderolabs/extensions/pkgs/container/iscsi-tools) | `v0.2.0` | This system extension provides iscsi-tools. | | [mdadm](storage/mdadm) | :warning: deprecated | [ghcr.io/siderolabs/mdadm](https://github.com/siderolabs/extensions/pkgs/container/mdadm) | `v4.6` | This system extension is no-op, for backward compatibility. | +| [mergerfs](storage/mergerfs) | :white_large_square: contrib | [ghcr.io/siderolabs/mergerfs](https://github.com/siderolabs/extensions/pkgs/container/mergerfs) | `2.42.0` | This system extension provides mergerfs, a FUSE union filesystem, as a host service that mounts one pool per /etc/mergerfs/*.ini file. | | [multipath-tools](storage/multipath-tools) | :yellow_square: extra | [ghcr.io/siderolabs/multipath-tools](https://github.com/siderolabs/extensions/pkgs/container/multipath-tools) | `v0.1.0` | This system extension provides multipath-tools. | | [nfs-server](storage/nfs-server) | :white_large_square: contrib | [ghcr.io/siderolabs/nfs-server](https://github.com/siderolabs/extensions/pkgs/container/nfs-server) | `v0.1.0` | This system extension provides the nfs-utils server-side daemons, turning the node into an NFS server serving NFSv3, v4.0, v4.1 and v4.2. exportfs syncs the kernel export table from /etc/exports. rpc.mountd services NFSv3 mount requests and the kernel's export upcalls. nfsdcld tracks NFSv4 clients so they can reclaim state after a server restart. rpc.nfsd starts the kernel NFS server threads. Requires the nfsd kernel module from the `nfsd` extension, and the `nfs-utils` extension for rpcbind and rpc.statd. | | [nfs-utils](storage/nfs-utils) | :white_large_square: contrib | [ghcr.io/siderolabs/nfs-utils](https://github.com/siderolabs/extensions/pkgs/container/nfs-utils) | `v0.1.1` | This system extension provides rpcbind and rpc.statd for NFSv3 file locking support. rpcbind is a server that converts RPC program numbers into universal addresses. rpc.statd is the NSM (Network Status Monitor) service daemon that notifies NFS peers of restarts. These services are required for NFSv3 mounts with file locking support. | diff --git a/storage/mergerfs/README.md b/storage/mergerfs/README.md new file mode 100644 index 00000000..b01a169d --- /dev/null +++ b/storage/mergerfs/README.md @@ -0,0 +1,104 @@ +# mergerfs + +This extension provides [mergerfs](https://github.com/trapexit/mergerfs), a FUSE union +filesystem that pools several filesystems ("branches") into one mount. The `ext-mergerfs` +host service runs one mergerfs process per config file in `/etc/mergerfs`. + +## Installation + +See [Installing Extensions](https://github.com/siderolabs/extensions#installing-extensions). + +Requires Talos v1.14.0 or newer for host-mode extension services. FUSE is built into the +Talos kernel, so no kernel module or other extension is needed. + +## Usage + +Describe each pool in an `EtcFileConfig` document. mergerfs reads all of its options from +the file as `key=value` lines: + +```yaml +apiVersion: v1alpha1 +kind: EtcFileConfig +name: mergerfs/media.ini +mode: 0o644 +contents: | + branches=/var/mnt/disk1:/var/mnt/disk2 + mountpoint=/var/mnt/media + fsname=media + category.create=pfrd + cache.files=partial + # wait for the branches to be mounted before pooling them + branches-mount-timeout=300 + branches-mount-timeout-fail=true +``` + +Branches and mountpoints have to live under `/var`, the writable part of the Talos +filesystem: user volumes are mounted at `/var/mnt/`, and ZFS datasets need a +`mountpoint` under `/var`. The mountpoint directory must exist (create it once from a +privileged debug pod) and must not be a symlink. `allow_other` is set automatically since +mergerfs runs as root. Everything else is covered by the +[mergerfs options](https://trapexit.github.io/mergerfs/latest/config/options/). + +### Exporting over NFS + +A pool can be exported with the `nfs-server` extension like any other directory. mergerfs +recommends `inodecalc=path-hash` and `never-forget-nodes=true` in the pool config, leaving +`lazy-umount-mountpoint` at its default, and an explicit `fsid=` on the export because +FUSE filesystems share a device number. The `mp` export option keeps the export inactive +until the pool is mounted: + +```text +/var/mnt/media 10.0.0.0/24(rw,sync,no_subtree_check,mp,fsid=0b6a5c1e-3c5b-4bd8-9e0d-2e8b6f1c7a42) +``` + +## How It Works + +The service reads `/etc/mergerfs` when it starts. When a file is added, changed or removed, +or a mergerfs process exits, it stops every pool and exits, and Talos restarts it five +seconds later with the current configuration; the mount a crashed mergerfs leaves behind is +detached first. A file without `mountpoint=`, or two files with the same mountpoint, keep +the service from starting until fixed, and the log says which. + +### Waiting for branches + +`ext-mergerfs` starts as soon as `/var` is mounted, which can be before the branches are, +for example before the `zfs` extension has imported its pools. `branches-mount-timeout` +makes mergerfs wait until every branch is a mount of its own, and +`branches-mount-timeout-fail=true` makes it exit rather than mount a partial pool. Branches +that are plain directories on the same filesystem as the mountpoint need a +`.mergerfs.branch` marker file instead. + +While waiting, mergerfs tries to run `mount` for each missing branch and aborts when there +is no such command, as on Talos. The service then restarts every five seconds, so the pool +still comes up once the branches are mounted; until then the log shows one +`mergerfs exited: signal: aborted` line per attempt. + +### Shutdown + +On reboot, shutdown, reset and upgrade, the extension's pre-shutdown hook unmounts every +pool while the services that own the branches (such as `ext-zfs-service`) and the ones +serving the pools (such as `ext-nfs-server`) are still running, so nothing races mergerfs +for the branches. A pool that has not unmounted after 20 seconds is lazily detached. The +hook never fails, since a failed hook aborts the whole sequence. + +## Testing + +Once the node is up, `talosctl service ext-mergerfs` is `Running` and the pools show up in +the service log and in the mount table: + +```bash +$ talosctl logs ext-mergerfs +10.5.0.3: mergerfs-supervisor: media.ini: started mergerfs on /var/mnt/media +$ talosctl mounts | grep /var/mnt/media +10.5.0.3 media 42.95 0.00 42.95 0.00% /var/mnt/media +``` + +A pod with a `hostPath` volume on the mountpoint and `mountPropagation: HostToContainer` +sees the files of every branch. After `talosctl reboot` the log shows +`draining: stopping all mounts` before the other services stop, and the pools are mounted +again after boot. + +## References + +- [mergerfs documentation](https://trapexit.github.io/mergerfs/) +- [mergerfs and NFS](https://trapexit.github.io/mergerfs/latest/remote_filesystems/) diff --git a/storage/mergerfs/manifest.yaml.tmpl b/storage/mergerfs/manifest.yaml.tmpl new file mode 100644 index 00000000..1de40af9 --- /dev/null +++ b/storage/mergerfs/manifest.yaml.tmpl @@ -0,0 +1,11 @@ +version: v1alpha1 +metadata: + name: mergerfs + version: "{{ .VERSION }}" + author: Micke Lisinge + description: | + [{{ .TIER }}] This system extension provides mergerfs, a FUSE union filesystem, as a host + service that mounts one pool per /etc/mergerfs/*.ini file. + compatibility: + talos: + version: ">= v1.14.0" diff --git a/storage/mergerfs/mergerfs.yaml b/storage/mergerfs/mergerfs.yaml new file mode 100644 index 00000000..7a9f4233 --- /dev/null +++ b/storage/mergerfs/mergerfs.yaml @@ -0,0 +1,15 @@ +name: mergerfs +runnerMode: host +depends: + # /var (and the user volumes and ZFS datasets under /var/mnt) is mounted. + - service: cri +container: + entrypoint: /usr/local/libexec/mergerfs-supervisor +restart: always +# Unmount every pool while the services that own the branches (ZFS, NFS) +# still run; without it they race mergerfs for the branches at shutdown. +preShutdown: + entrypoint: /usr/local/libexec/mergerfs-supervisor + args: + - pre-shutdown + timeout: 30s diff --git a/storage/mergerfs/mounts/cmd/mergerfs-supervisor/main.go b/storage/mergerfs/mounts/cmd/mergerfs-supervisor/main.go new file mode 100644 index 00000000..fdc18003 --- /dev/null +++ b/storage/mergerfs/mounts/cmd/mergerfs-supervisor/main.go @@ -0,0 +1,51 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +// Command mergerfs-supervisor runs one mergerfs per /etc/mergerfs/*.ini. +// +// Without arguments it supervises the mounts. `mergerfs-supervisor pre-shutdown` +// is the Talos pre-shutdown hook: it makes the running supervisor unmount +// everything while the services owning the branches (ZFS, NFS) still run. +package main + +import ( + "context" + "log" + "os" + "os/signal" + "path/filepath" + "syscall" + + "github.com/siderolabs/extensions/storage/mergerfs/mounts/server" +) + +func main() { + log.SetFlags(0) + log.SetPrefix("mergerfs-supervisor: ") + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + runtime := server.NewOSRuntime() + config := server.DefaultConfig() + + switch { + case len(os.Args) == 1: + if err := server.Run(ctx, runtime, config); err != nil { + log.Print(err) + + os.Exit(1) + } + case len(os.Args) == 2 && os.Args[1] == "pre-shutdown": + // A failed hook makes Talos abort the reboot, shutdown or upgrade; + // a mount that is still there is the lesser problem, so only log. + if err := server.PreShutdown(ctx, runtime, config); err != nil { + log.Printf("pre-shutdown: %v", err) + } + default: + log.Printf("usage: %s [pre-shutdown]", filepath.Base(os.Args[0])) + + os.Exit(1) + } +} diff --git a/storage/mergerfs/mounts/go.mod b/storage/mergerfs/mounts/go.mod new file mode 100644 index 00000000..84696044 --- /dev/null +++ b/storage/mergerfs/mounts/go.mod @@ -0,0 +1,3 @@ +module github.com/siderolabs/extensions/storage/mergerfs/mounts + +go 1.26.8 diff --git a/storage/mergerfs/mounts/mounts.go b/storage/mergerfs/mounts/mounts.go new file mode 100644 index 00000000..fc6e963f --- /dev/null +++ b/storage/mergerfs/mounts/mounts.go @@ -0,0 +1,151 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +// Package mounts inspects the mount table for mergerfs mounts and detaches the +// ones left behind by a mergerfs process that died. +package mounts + +import ( + "bytes" + "fmt" + "os" + "strings" + "syscall" +) + +const ( + mountinfoPath = "/proc/self/mountinfo" + + // Filesystem type mergerfs mounts are recorded as. + fsType = "fuse.mergerfs" + + // maxStaleLayers bounds DetachStale's loop; each restart of a crashed + // mergerfs adds at most one layer. + maxStaleLayers = 64 +) + +// Mounter provides the filesystem operations required by DetachStale. +type Mounter interface { + ReadMountinfo() ([]byte, error) + Unmount(target string, flags int) error +} + +// OSMounter performs mount operations using the host operating system. +type OSMounter struct{} + +// ReadMountinfo returns the mount table of the calling process. +func (OSMounter) ReadMountinfo() ([]byte, error) { + return os.ReadFile(mountinfoPath) +} + +// Unmount unmounts a filesystem. +func (OSMounter) Unmount(target string, flags int) error { + return syscall.Unmount(target, flags) +} + +// IsMounted reports whether a mergerfs mount is what is visible at mountpoint. +// +// The mount table records the path a mount was made at, so mountpoint has to +// be that path: cleaned and without symlinks. It is never looked up on disk, +// which would hang on a wedged FUSE mount. +func IsMounted(mounter Mounter, mountpoint string) (bool, error) { + mountinfo, err := mounter.ReadMountinfo() + if err != nil { + return false, fmt.Errorf("read mount table: %w", err) + } + + return IsMergerfsMount(mountinfo, mountpoint), nil +} + +// DetachStale lazily detaches the mergerfs mounts stacked at mountpoint (see +// IsMounted for the form it takes) and returns how many there were. +// +// A mergerfs process that dies without unmounting leaves its mount behind, +// answering ENOTCONN; mounting a new instance on top of it would stack a second +// mount. Only the mount visible at mountpoint is considered, and only while it +// is a `fuse.mergerfs` mount: a different filesystem mounted over a stale +// mergerfs mount is neither detached nor looked under. +func DetachStale(mounter Mounter, mountpoint string) (int, error) { + detached := 0 + + for range maxStaleLayers { + mountinfo, err := mounter.ReadMountinfo() + if err != nil { + return detached, fmt.Errorf("read mount table: %w", err) + } + + if !IsMergerfsMount(mountinfo, mountpoint) { + return detached, nil + } + + if err = mounter.Unmount(mountpoint, syscall.MNT_DETACH); err != nil { + return detached, fmt.Errorf("detach %s: %w", mountpoint, err) + } + + detached++ + } + + return detached, fmt.Errorf("detach %s: still mounted after %d detaches", mountpoint, detached) +} + +// IsMergerfsMount reports whether the mount visible at path is a fuse.mergerfs mount. +// +// mountinfo lists mounts in the order they were made, so of several mounts at +// the same path the last entry is the one on top. +func IsMergerfsMount(mountinfo []byte, path string) bool { + fstype, ok := TopMount(mountinfo, path) + + return ok && fstype == fsType +} + +// TopMount returns the filesystem type of the mount visible at path. +func TopMount(mountinfo []byte, path string) (fstype string, ok bool) { + for line := range bytes.Lines(mountinfo) { + // 36 35 98:0 /mnt1 /mnt2 rw,noatime master:1 - fuse.mergerfs pool rw,... + pre, post, found := strings.Cut(strings.TrimRight(string(line), "\n"), " - ") + if !found { + continue + } + + fields := strings.Fields(pre) + postFields := strings.Fields(post) + + if len(fields) < 5 || len(postFields) < 1 { + continue + } + + if unescape(fields[4]) == path { + fstype, ok = postFields[0], true + } + } + + return fstype, ok +} + +// unescape decodes the octal escapes (\040 for space, \011, \012, \134) the kernel +// uses for whitespace and backslashes in mountinfo paths. +func unescape(field string) string { + if !strings.Contains(field, `\`) { + return field + } + + var builder strings.Builder + + for i := 0; i < len(field); i++ { + if field[i] == '\\' && i+3 < len(field) && isOctal(field[i+1]) && isOctal(field[i+2]) && isOctal(field[i+3]) { + builder.WriteByte((field[i+1]-'0')<<6 | (field[i+2]-'0')<<3 | (field[i+3] - '0')) + i += 3 + + continue + } + + builder.WriteByte(field[i]) + } + + return builder.String() +} + +func isOctal(c byte) bool { + return c >= '0' && c <= '7' +} diff --git a/storage/mergerfs/mounts/mounts_test.go b/storage/mergerfs/mounts/mounts_test.go new file mode 100644 index 00000000..1444e9a2 --- /dev/null +++ b/storage/mergerfs/mounts/mounts_test.go @@ -0,0 +1,158 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +package mounts_test + +import ( + "errors" + "slices" + "strings" + "syscall" + "testing" + + "github.com/siderolabs/extensions/storage/mergerfs/mounts" +) + +const mountinfo = `22 1 0:21 / / ro,relatime - squashfs /dev/loop0 ro +98 30 0:52 / /var/mnt/disk1 rw,noatime shared:40 - zfs disk1 rw,xattr,posixacl +120 30 0:61 / /var/mnt/pool rw,relatime shared:55 - fuse.mergerfs pool rw,user_id=0,group_id=0,allow_other +121 30 0:62 / /var/mnt/with\040space rw,relatime shared:56 - fuse.mergerfs spaced rw +122 30 0:63 / /var/mnt/other-fuse rw,relatime shared:57 - fuse.sshfs remote rw +123 30 0:64 / /var/mnt/covered rw,relatime shared:58 - fuse.mergerfs stale rw +124 30 0:65 / /var/mnt/covered rw,relatime shared:59 - tmpfs tmpfs rw +125 30 0:66 / /var/mnt/stacked rw,relatime shared:60 - fuse.mergerfs stale rw +126 30 0:67 / /var/mnt/stacked rw,relatime shared:61 - fuse.mergerfs stale rw +` + +type call struct { + target string + flags int +} + +type fakeMounter struct { + mountinfo string + readErr error + unmounts []call + unmountErr error + // sticky keeps the mount table unchanged after an unmount. + sticky bool +} + +func (m *fakeMounter) ReadMountinfo() ([]byte, error) { + return []byte(m.mountinfo), m.readErr +} + +// Unmount drops the last mountinfo entry for target, like the kernel does. +func (m *fakeMounter) Unmount(target string, flags int) error { + m.unmounts = append(m.unmounts, call{target: target, flags: flags}) + + if m.unmountErr != nil { + return m.unmountErr + } + + if m.sticky { + return nil + } + + lines := strings.Split(strings.TrimSuffix(m.mountinfo, "\n"), "\n") + for i := len(lines) - 1; i >= 0; i-- { + if strings.Fields(lines[i])[4] == target { + m.mountinfo = strings.Join(slices.Delete(lines, i, i+1), "\n") + "\n" + + break + } + } + + return nil +} + +func TestIsMergerfsMount(t *testing.T) { + t.Parallel() + + for path, want := range map[string]bool{ + "/var/mnt/pool": true, + "/var/mnt/with space": true, + "/var/mnt/stacked": true, + "/var/mnt/disk1": false, // not mergerfs: never ours to unmount + "/var/mnt/other-fuse": false, + "/var/mnt/covered": false, // a stale mergerfs mount under something else is left alone + "/var/mnt/missing": false, + } { + if got := mounts.IsMergerfsMount([]byte(mountinfo), path); got != want { + t.Errorf("IsMergerfsMount(%q) = %v, want %v", path, got, want) + } + } +} + +func TestDetachStaleDetachesOnlyMergerfsMounts(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + mountpoint string + detached int + }{ + {mountpoint: "/var/mnt/pool", detached: 1}, + {mountpoint: "/var/mnt/stacked", detached: 2}, + {mountpoint: "/var/mnt/covered", detached: 0}, + {mountpoint: "/var/mnt/disk1", detached: 0}, + {mountpoint: "/var/mnt/missing", detached: 0}, + } { + mounter := &fakeMounter{mountinfo: mountinfo} + + detached, err := mounts.DetachStale(mounter, tc.mountpoint) + if err != nil { + t.Fatalf("DetachStale(%q) error = %v", tc.mountpoint, err) + } + + if detached != tc.detached { + t.Fatalf("DetachStale(%q) = %v, want %v", tc.mountpoint, detached, tc.detached) + } + + var want []call + for range tc.detached { + want = append(want, call{target: tc.mountpoint, flags: syscall.MNT_DETACH}) + } + + if !slices.Equal(mounter.unmounts, want) { + t.Fatalf("DetachStale(%q) unmounts = %v, want %v", tc.mountpoint, mounter.unmounts, want) + } + + if mounts.IsMergerfsMount([]byte(mounter.mountinfo), tc.mountpoint) { + t.Fatalf("DetachStale(%q) left a mergerfs mount on top", tc.mountpoint) + } + } +} + +func TestDetachStaleReturnsErrors(t *testing.T) { + t.Parallel() + + readErr := errors.New("read failed") + if _, err := mounts.DetachStale(&fakeMounter{readErr: readErr}, "/var/mnt/pool"); !errors.Is(err, readErr) { + t.Fatalf("DetachStale() error = %v, want %v", err, readErr) + } + + if _, err := mounts.DetachStale(&fakeMounter{mountinfo: mountinfo, unmountErr: syscall.EINVAL}, "/var/mnt/pool"); !errors.Is(err, syscall.EINVAL) { + t.Fatalf("DetachStale() error = %v, want EINVAL", err) + } + + // An unmount that never takes effect must not loop forever. + sticky := &fakeMounter{mountinfo: mountinfo, sticky: true} + if _, err := mounts.DetachStale(sticky, "/var/mnt/pool"); err == nil { + t.Fatal("DetachStale() with a sticky mount returned no error") + } +} + +func TestIsMounted(t *testing.T) { + t.Parallel() + + mounted, err := mounts.IsMounted(&fakeMounter{mountinfo: mountinfo}, "/var/mnt/pool") + if err != nil || !mounted { + t.Fatalf("IsMounted(pool) = %v, %v; want true", mounted, err) + } + + mounted, err = mounts.IsMounted(&fakeMounter{mountinfo: mountinfo}, "/var/mnt/covered") + if err != nil || mounted { + t.Fatalf("IsMounted(covered) = %v, %v; want false", mounted, err) + } +} diff --git a/storage/mergerfs/mounts/server/drain.go b/storage/mergerfs/mounts/server/drain.go new file mode 100644 index 00000000..c2c2f2de --- /dev/null +++ b/storage/mergerfs/mounts/server/drain.go @@ -0,0 +1,160 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +package server + +import ( + "context" + "errors" + "fmt" + "io/fs" + "maps" + "os" + "path/filepath" + "slices" + "time" +) + +const ( + // DefaultStateDir holds the drain request. + DefaultStateDir = "/run/mergerfs" + + // DefaultDrainMaxAge is how long a drain request stays in force. Talos + // stops the services within seconds of the pre-shutdown hooks; a request + // still around after this belongs to a shutdown that was aborted, and the + // pools come back on their own. + DefaultDrainMaxAge = 5 * time.Minute + + drainFile = "drain" +) + +// Drain is the drain request file the pre-shutdown hook leaves for the +// supervisor. Its modification time is the time of the request. +type Drain struct { + Dir string + MaxAge time.Duration +} + +// Request writes the drain request. +func (d Drain) Request() error { + if err := os.MkdirAll(d.Dir, 0o755); err != nil { + return err + } + + path := filepath.Join(d.Dir, drainFile) + + if err := os.WriteFile(path, nil, 0o644); err != nil { + return err + } + + now := time.Now() + + return os.Chtimes(path, now, now) +} + +// Active reports whether a drain request is in force, removing one that has expired. +func (d Drain) Active(now time.Time) (bool, error) { + path := filepath.Join(d.Dir, drainFile) + + info, err := os.Stat(path) + if errors.Is(err, fs.ErrNotExist) { + return false, nil + } + + if err != nil { + return false, err + } + + if now.Sub(info.ModTime()) <= d.MaxAge { + return true, nil + } + + if err = os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) { + return false, err + } + + return false, nil +} + +// PreShutdown is the pre-shutdown hook: it requests a drain and waits until no +// configured mountpoint carries a mergerfs mount any more. Mounts still there +// after HookWait, for example because the supervisor is not running, are +// lazily detached here. +// +// The returned error describes what went wrong; the caller decides whether it +// fails the hook. Talos aborts the whole reboot on a failed hook, which is +// worse than a late unmount, so the command never does. +func PreShutdown(ctx context.Context, runtime Runtime, config Config) error { + config = normalizeConfig(config) + + if err := runtime.RequestDrain(); err != nil { + return fmt.Errorf("request drain: %w", err) + } + + configs, err := runtime.ReadConfigs(config.Dir) + if err != nil { + return fmt.Errorf("read %s: %w", config.Dir, err) + } + + mountpoints := map[string]string{} // by config name + + for name, content := range configs { + if mountpoint := Mountpoint(content); mountpoint != "" { + mountpoints[name] = mountpoint + } + } + + started := time.Now() + deadline := time.NewTimer(config.HookWait) + ticker := time.NewTicker(config.HookPoll) + + defer deadline.Stop() + defer ticker.Stop() + + for { + // A mountpoint that cannot be checked counts as still mounted; the + // error is reported once, with the final result. + var ( + remaining []string + errs []error + ) + + for _, name := range slices.Sorted(maps.Keys(mountpoints)) { + mounted, err := runtime.IsMounted(mountpoints[name]) + if err != nil { + errs = append(errs, fmt.Errorf("%s: %w", name, err)) + } + + if mounted || err != nil { + remaining = append(remaining, name) + } + } + + if len(remaining) == 0 { + config.Logger.Printf("drained %d mount(s) in %s", len(mountpoints), time.Since(started).Round(time.Millisecond)) + + return nil + } + + select { + case <-ticker.C: + continue + case <-ctx.Done(): + case <-deadline.C: + } + + for _, name := range remaining { + detached, err := runtime.DetachStale(mountpoints[name]) + if detached > 0 { + config.Logger.Printf("%s: detached %d leftover mount(s) at %s", name, detached, mountpoints[name]) + } + + if err != nil { + errs = append(errs, fmt.Errorf("%s: %w", name, err)) + } + } + + return errors.Join(append(errs, fmt.Errorf("%d mount(s) still there after %s: %v", len(remaining), time.Since(started).Round(time.Millisecond), remaining))...) + } +} diff --git a/storage/mergerfs/mounts/server/drain_test.go b/storage/mergerfs/mounts/server/drain_test.go new file mode 100644 index 00000000..eb23aa76 --- /dev/null +++ b/storage/mergerfs/mounts/server/drain_test.go @@ -0,0 +1,128 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +package server_test + +import ( + "errors" + "os" + "path/filepath" + "slices" + "testing" + "time" + + "github.com/siderolabs/extensions/storage/mergerfs/mounts/server" +) + +func TestDrainFile(t *testing.T) { + t.Parallel() + + drain := server.Drain{Dir: filepath.Join(t.TempDir(), "state"), MaxAge: time.Minute} + + now := time.Now() + + active, err := drain.Active(now) + if err != nil || active { + t.Fatalf("Active() before any request = %v, %v; want false", active, err) + } + + if err = drain.Request(); err != nil { + t.Fatalf("Request() error = %v", err) + } + + active, err = drain.Active(now) + if err != nil || !active { + t.Fatalf("Active() after a request = %v, %v; want true", active, err) + } + + // An aborted shutdown: the request expires and is removed. + active, err = drain.Active(now.Add(2 * time.Minute)) + if err != nil || active { + t.Fatalf("Active() after MaxAge = %v, %v; want false", active, err) + } + + if _, err = os.Stat(filepath.Join(drain.Dir, "drain")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("expired request was not removed: %v", err) + } + + // A repeated request is a fresh one. + if err = drain.Request(); err != nil { + t.Fatalf("Request() error = %v", err) + } + + active, err = drain.Active(now) + if err != nil || !active { + t.Fatalf("Active() after a repeated request = %v, %v; want true", active, err) + } +} + +func TestPreShutdownWaitsForTheSupervisor(t *testing.T) { + t.Parallel() + + runtime := newFakeRuntime(t, map[string]string{"pool.ini": poolConfig, "pool-slow.ini": poolSlowConfig}) + start(t, runtime) + + eventually(t, "both mounts", func() bool { + _, _, _, mounted := runtime.snapshot() + + return len(mounted) == 2 + }) + + if err := server.PreShutdown(t.Context(), runtime, testConfig()); err != nil { + t.Fatalf("PreShutdown() error = %v", err) + } + + _, unmounts, stops, mounted := runtime.snapshot() + if len(stops) != 2 || len(mounted) != 0 || len(unmounts) != 0 { + t.Fatalf("expected the supervisor to stop both mounts: stops %v, unmounts %v, mounted %v", stops, unmounts, mounted) + } + + if runtime.drainCalls != 1 { + t.Fatalf("drain requested %d times, want 1", runtime.drainCalls) + } +} + +func TestPreShutdownDetachesLeftoversWithoutASupervisor(t *testing.T) { + t.Parallel() + + runtime := newFakeRuntime(t, map[string]string{"pool.ini": poolConfig, "pool-slow.ini": poolSlowConfig}) + runtime.mounted["/var/mnt/pool"] = 1 + + err := server.PreShutdown(t.Context(), runtime, testConfig()) + if err == nil { + t.Fatal("PreShutdown() with nobody to drain returned no error") + } + + _, unmounts, _, mounted := runtime.snapshot() + if !slices.Equal(unmounts, []string{"/var/mnt/pool"}) || len(mounted) != 0 { + t.Fatalf("expected the leftover to be detached: unmounts %v, mounted %v", unmounts, mounted) + } +} + +func TestPreShutdownWithoutMountsReturnsAtOnce(t *testing.T) { + t.Parallel() + + runtime := newFakeRuntime(t, map[string]string{"pool.ini": poolConfig}) + + started := time.Now() + + if err := server.PreShutdown(t.Context(), runtime, testConfig()); err != nil { + t.Fatalf("PreShutdown() error = %v", err) + } + + if elapsed := time.Since(started); elapsed > testConfig().HookWait { + t.Fatalf("PreShutdown() waited %s with nothing mounted", elapsed) + } +} + +func TestPreShutdownReportsAFailedRequest(t *testing.T) { + t.Parallel() + + runtime := newFakeRuntime(t, map[string]string{"pool.ini": poolConfig}) + runtime.drainErr = errors.New("read-only file system") + + if err := server.PreShutdown(t.Context(), runtime, testConfig()); !errors.Is(err, runtime.drainErr) { + t.Fatalf("PreShutdown() error = %v, want %v", err, runtime.drainErr) + } +} diff --git a/storage/mergerfs/mounts/server/osruntime.go b/storage/mergerfs/mounts/server/osruntime.go new file mode 100644 index 00000000..8ad8c583 --- /dev/null +++ b/storage/mergerfs/mounts/server/osruntime.go @@ -0,0 +1,159 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +package server + +import ( + "context" + "errors" + "io/fs" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "syscall" + "time" + + "github.com/siderolabs/extensions/storage/mergerfs/mounts" +) + +// MergerfsPath is where the extension installs the mergerfs binary. +const MergerfsPath = "/usr/local/bin/mergerfs" + +// OSRuntime runs mergerfs on the Talos host. +type OSRuntime struct { + drain Drain +} + +// NewOSRuntime returns a Runtime backed by host operating-system primitives. +func NewOSRuntime() OSRuntime { + return OSRuntime{ + drain: Drain{Dir: DefaultStateDir, MaxAge: DefaultDrainMaxAge}, + } +} + +// ReadConfigs implements Runtime. +func (OSRuntime) ReadConfigs(dir string) (map[string][]byte, error) { + entries, err := os.ReadDir(dir) + if errors.Is(err, fs.ErrNotExist) { + return map[string][]byte{}, nil + } + + if err != nil { + return nil, err + } + + configs := map[string][]byte{} + + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".ini") { + continue + } + + content, readErr := os.ReadFile(filepath.Join(dir, entry.Name())) + if readErr != nil { + return nil, readErr + } + + configs[entry.Name()] = content + } + + return configs, nil +} + +// Start implements Runtime. +func (OSRuntime) Start(path string) (Process, error) { + cmd := exec.Command(MergerfsPath, "-f", "-o", "config="+path) + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + // If the supervisor dies, mergerfs gets SIGTERM and unmounts on its own; + // the next supervisor clears whatever is left before remounting. + cmd.SysProcAttr = &syscall.SysProcAttr{Pdeathsig: syscall.SIGTERM} + + if err := cmd.Start(); err != nil { + return nil, err + } + + process := &osProcess{cmd: cmd, done: make(chan struct{})} + go process.wait() + + return process, nil +} + +// DetachStale implements Runtime. +func (OSRuntime) DetachStale(mountpoint string) (int, error) { + return mounts.DetachStale(mounts.OSMounter{}, mountpoint) +} + +// IsMounted implements Runtime. +func (OSRuntime) IsMounted(mountpoint string) (bool, error) { + return mounts.IsMounted(mounts.OSMounter{}, mountpoint) +} + +// Draining implements Runtime. +func (r OSRuntime) Draining(now time.Time) (bool, error) { + return r.drain.Active(now) +} + +// RequestDrain implements Runtime. +func (r OSRuntime) RequestDrain() error { + return r.drain.Request() +} + +type osProcess struct { + cmd *exec.Cmd + done chan struct{} + + mu sync.Mutex + err error +} + +func (process *osProcess) wait() { + err := process.cmd.Wait() + + process.mu.Lock() + process.err = err + process.mu.Unlock() + + close(process.done) +} + +func (process *osProcess) Done() <-chan struct{} { + return process.done +} + +func (process *osProcess) Err() error { + process.mu.Lock() + defer process.mu.Unlock() + + return process.err +} + +// Stop sends SIGTERM, on which mergerfs detaches its mount and exits, and +// falls back to SIGKILL when ctx expires. It does not wait for the kill to +// take effect: a process stuck in the kernel would hold up the supervisor +// (the same choice as the nfs-server extension makes). +func (process *osProcess) Stop(ctx context.Context) error { + select { + case <-process.done: + return nil + default: + } + + if err := process.cmd.Process.Signal(syscall.SIGTERM); err != nil && !errors.Is(err, os.ErrProcessDone) { + return err + } + + select { + case <-process.done: + return nil + case <-ctx.Done(): + if err := process.cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) { + return errors.Join(ctx.Err(), err) + } + + return ctx.Err() + } +} diff --git a/storage/mergerfs/mounts/server/server.go b/storage/mergerfs/mounts/server/server.go new file mode 100644 index 00000000..9ea184c7 --- /dev/null +++ b/storage/mergerfs/mounts/server/server.go @@ -0,0 +1,375 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +// Package server runs one mergerfs process per config file and stops them all +// when one of them exits or the configuration changes, so Talos restarts the +// service. +package server + +import ( + "bytes" + "context" + "errors" + "fmt" + "log" + "maps" + "path/filepath" + "slices" + "strings" + "sync" + "time" +) + +const ( + // DefaultConfigDir holds one `.ini` per mergerfs mount, supplied with EtcFileConfig. + DefaultConfigDir = "/etc/mergerfs" + + defaultPollInterval = time.Second + defaultStopTimeout = 8 * time.Second + defaultHookWait = 20 * time.Second + defaultHookPoll = 100 * time.Millisecond +) + +// Process is a running mergerfs instance. +type Process interface { + Done() <-chan struct{} + Err() error + // Stop asks the process to exit and returns once it has, or once ctx + // expires, in which case it has been killed. + Stop(context.Context) error +} + +// Runtime provides the host operations the supervisor and the pre-shutdown hook need. +type Runtime interface { + // ReadConfigs returns the contents of every `*.ini` file in dir, keyed by file name. + // A missing directory is not an error: it means no mounts are configured. + ReadConfigs(dir string) (map[string][]byte, error) + // Start launches `mergerfs -f -o config=`. + Start(path string) (Process, error) + // DetachStale lazily detaches the mergerfs mounts left at mountpoint and + // returns how many there were. + DetachStale(mountpoint string) (int, error) + // IsMounted reports whether a mergerfs mount is visible at mountpoint. + IsMounted(mountpoint string) (bool, error) + // Draining reports whether a drain request is in force. + Draining(now time.Time) (bool, error) + // RequestDrain asks the supervisor to unmount everything and keep it unmounted. + RequestDrain() error +} + +// Config controls supervision. +type Config struct { + Dir string + PollInterval time.Duration + // StopTimeout bounds stopping the mounts. It stays below Talos's + // ten-second host-process termination grace period. + StopTimeout time.Duration + // HookWait is how long the pre-shutdown hook waits for the supervisor to + // unmount everything before detaching what is left itself. + HookWait time.Duration + HookPoll time.Duration + Logger *log.Logger +} + +// DefaultConfig returns production settings. +func DefaultConfig() Config { + return Config{ + Dir: DefaultConfigDir, + PollInterval: defaultPollInterval, + StopTimeout: defaultStopTimeout, + HookWait: defaultHookWait, + HookPoll: defaultHookPoll, + Logger: log.Default(), + } +} + +func normalizeConfig(config Config) Config { + defaults := DefaultConfig() + + if config.Dir == "" { + config.Dir = defaults.Dir + } + + if config.PollInterval <= 0 { + config.PollInterval = defaults.PollInterval + } + + if config.StopTimeout <= 0 { + config.StopTimeout = defaults.StopTimeout + } + + if config.HookWait <= 0 { + config.HookWait = defaults.HookWait + } + + if config.HookPoll <= 0 { + config.HookPoll = defaults.HookPoll + } + + if config.Logger == nil { + config.Logger = defaults.Logger + } + + return config +} + +type mount struct { + name string + path string + mountpoint string + process Process +} + +type supervisor struct { + runtime Runtime + config Config + mounts []*mount +} + +// Run mounts every configured pool and supervises the mergerfs processes until +// one of them exits, the configuration changes, a drain is requested (see +// PreShutdown) or ctx is canceled. It stops every mount before returning, and +// Talos restarts the service. The returned error says why it stopped. +func Run(ctx context.Context, runtime Runtime, config Config) error { + config = normalizeConfig(config) + + configs, err := runtime.ReadConfigs(config.Dir) + if err != nil { + return fmt.Errorf("read %s: %w", config.Dir, err) + } + + mounts, err := plan(config.Dir, configs) + if err != nil { + return err + } + + if len(mounts) == 0 { + config.Logger.Printf("no config files in %s", config.Dir) + } + + s := &supervisor{runtime: runtime, config: config, mounts: mounts} + + ticker := time.NewTicker(config.PollInterval) + defer ticker.Stop() + + // A drain request keeps the pools unmounted until it expires. + for waited := false; ; waited = true { + draining, err := runtime.Draining(time.Now()) + if err != nil { + return fmt.Errorf("check drain request: %w", err) + } + + if !draining { + break + } + + if !waited { + config.Logger.Printf("drain request in force, waiting") + } + + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + } + } + + if err = s.startAll(); err != nil { + return errors.Join(err, s.stopAll()) + } + + for { + select { + case <-ctx.Done(): + return s.stopAll() + case <-ticker.C: + } + + if name, err := s.exited(); name != "" { + return errors.Join(fmt.Errorf("%s: mergerfs exited: %w", name, err), s.stopAll()) + } + + draining, err := runtime.Draining(time.Now()) + if err != nil { + return errors.Join(fmt.Errorf("check drain request: %w", err), s.stopAll()) + } + + if draining { + config.Logger.Printf("draining: stopping all mounts") + + return s.stopAll() + } + + // A transient read error must not unmount anything. + current, err := runtime.ReadConfigs(config.Dir) + if err != nil || maps.EqualFunc(current, configs, bytes.Equal) { + continue + } + + config.Logger.Printf("config changed, restarting") + + return s.stopAll() + } +} + +// plan turns the config files into mounts, sorted by name. +func plan(dir string, configs map[string][]byte) ([]*mount, error) { + var mounts []*mount + + owners := map[string]string{} + + for _, name := range slices.Sorted(maps.Keys(configs)) { + mountpoint := Mountpoint(configs[name]) + if mountpoint == "" { + return nil, fmt.Errorf("%s: no mountpoint= in config", name) + } + + if owner, taken := owners[mountpoint]; taken { + return nil, fmt.Errorf("%s: mountpoint %s is used by %s", name, mountpoint, owner) + } + + owners[mountpoint] = name + + mounts = append(mounts, &mount{ + name: name, + path: filepath.Join(dir, name), + mountpoint: mountpoint, + }) + } + + return mounts, nil +} + +func (s *supervisor) startAll() error { + for _, m := range s.mounts { + // Mounting over a leftover mergerfs mount would stack a second one on top. + if err := s.detachStale(m); err != nil { + return err + } + + process, err := s.runtime.Start(m.path) + if err != nil { + return fmt.Errorf("%s: start mergerfs: %w", m.name, err) + } + + m.process = process + + s.config.Logger.Printf("%s: started mergerfs on %s", m.name, m.mountpoint) + } + + return nil +} + +// exited returns the first mount whose mergerfs process has exited. +func (s *supervisor) exited() (string, error) { + for _, m := range s.mounts { + if m.process == nil { + continue + } + + select { + case <-m.process.Done(): + default: + continue + } + + err := m.process.Err() + if err == nil { + err = errors.New("exited without an error") + } + + return m.name, err + } + + return "", nil +} + +// stopAll stops every mount concurrently, bounded by StopTimeout. The service +// is stopping either way, so the stop gets its own budget. +func (s *supervisor) stopAll() error { + ctx, cancel := context.WithTimeout(context.Background(), s.config.StopTimeout) + defer cancel() + + var ( + wg sync.WaitGroup + mu sync.Mutex + errs []error + ) + + for _, m := range s.mounts { + wg.Go(func() { + if err := s.stop(ctx, m); err != nil { + mu.Lock() + errs = append(errs, fmt.Errorf("%s: %w", m.name, err)) + mu.Unlock() + } + }) + } + + wg.Wait() + + return errors.Join(errs...) +} + +func (s *supervisor) stop(ctx context.Context, m *mount) error { + var errs []error + + if m.process != nil { + if err := m.process.Stop(ctx); err != nil { + errs = append(errs, fmt.Errorf("stop mergerfs: %w", err)) + } + + m.process = nil + } + + // mergerfs detaches its own mount on SIGTERM; this covers a SIGKILL. + if err := s.detachStale(m); err != nil { + errs = append(errs, err) + } + + return errors.Join(errs...) +} + +// detachStale removes the mergerfs mounts a dead instance left at the +// mountpoint, which answer ENOTCONN. +func (s *supervisor) detachStale(m *mount) error { + detached, err := s.runtime.DetachStale(m.mountpoint) + if detached > 0 { + s.config.Logger.Printf("%s: unmounted %d leftover mount(s) at %s", m.name, detached, m.mountpoint) + } + + return err +} + +// Mountpoint returns the mountpoint declared by a mergerfs config file, cleaned. +// +// mergerfs reads `key=value` lines, ignoring blank lines and `#` comments, and +// accepts both `mountpoint` and its alias `mount`. Options are applied in +// order, so the last declaration wins. +func Mountpoint(config []byte) string { + var mountpoint string + + for line := range bytes.Lines(config) { + text := strings.TrimSpace(string(line)) + if text == "" || strings.HasPrefix(text, "#") { + continue + } + + key, value, ok := strings.Cut(text, "=") + if !ok { + continue + } + + switch strings.TrimSpace(key) { + case "mountpoint", "mount": + mountpoint = strings.TrimSpace(value) + } + } + + if mountpoint == "" { + return "" + } + + return filepath.Clean(mountpoint) +} diff --git a/storage/mergerfs/mounts/server/server_test.go b/storage/mergerfs/mounts/server/server_test.go new file mode 100644 index 00000000..566969ba --- /dev/null +++ b/storage/mergerfs/mounts/server/server_test.go @@ -0,0 +1,543 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. + +package server_test + +import ( + "context" + "errors" + "io" + "log" + "maps" + "os" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/siderolabs/extensions/storage/mergerfs/mounts/server" +) + +type fakeProcess struct { + path string + mountpoint string + done chan struct{} + once sync.Once + + mu sync.Mutex + err error +} + +func (process *fakeProcess) Done() <-chan struct{} { return process.done } + +func (process *fakeProcess) Err() error { + process.mu.Lock() + defer process.mu.Unlock() + + return process.err +} + +func (process *fakeProcess) exit(err error) { + process.mu.Lock() + process.err = err + process.mu.Unlock() + + process.once.Do(func() { close(process.done) }) +} + +type fakeRuntime struct { + t *testing.T + + mu sync.Mutex + configs map[string][]byte + running map[string]*fakeProcess // by config path + mounted map[string]int // stacked mergerfs mounts by mountpoint + starts []string + unmounts []string + stopCalls []string + draining bool + drainErr error + drainCalls int +} + +func newFakeRuntime(t *testing.T, configs map[string]string) *fakeRuntime { + t.Helper() + + runtime := &fakeRuntime{ + t: t, + configs: map[string][]byte{}, + running: map[string]*fakeProcess{}, + mounted: map[string]int{}, + } + + for name, content := range configs { + runtime.configs[name] = []byte(content) + } + + return runtime +} + +func (runtime *fakeRuntime) ReadConfigs(string) (map[string][]byte, error) { + runtime.mu.Lock() + defer runtime.mu.Unlock() + + return maps.Clone(runtime.configs), nil +} + +func (runtime *fakeRuntime) Start(path string) (server.Process, error) { + runtime.mu.Lock() + defer runtime.mu.Unlock() + + mountpoint := mountpointFor(runtime.configs, path) + + if runtime.mounted[mountpoint] > 0 { + runtime.t.Errorf("start %s over an existing mount", path) + } + + process := &fakeProcess{path: path, mountpoint: mountpoint, done: make(chan struct{})} + runtime.running[path] = process + runtime.starts = append(runtime.starts, path) + runtime.mounted[mountpoint]++ + + return &stoppable{fakeProcess: process, runtime: runtime}, nil +} + +func (runtime *fakeRuntime) DetachStale(mountpoint string) (int, error) { + runtime.mu.Lock() + defer runtime.mu.Unlock() + + detached := runtime.mounted[mountpoint] + delete(runtime.mounted, mountpoint) + + for range detached { + runtime.unmounts = append(runtime.unmounts, mountpoint) + } + + return detached, nil +} + +func (runtime *fakeRuntime) IsMounted(mountpoint string) (bool, error) { + runtime.mu.Lock() + defer runtime.mu.Unlock() + + return runtime.mounted[mountpoint] > 0, nil +} + +func (runtime *fakeRuntime) Draining(time.Time) (bool, error) { + runtime.mu.Lock() + defer runtime.mu.Unlock() + + return runtime.draining, runtime.drainErr +} + +func (runtime *fakeRuntime) RequestDrain() error { + runtime.mu.Lock() + defer runtime.mu.Unlock() + + runtime.drainCalls++ + + if runtime.drainErr != nil { + return runtime.drainErr + } + + runtime.draining = true + + return nil +} + +func (runtime *fakeRuntime) setDraining(draining bool) { + runtime.mu.Lock() + defer runtime.mu.Unlock() + + runtime.draining = draining +} + +// crash kills the mergerfs process started from path with SIGKILL semantics: +// the process dies, its mount stays. +func (runtime *fakeRuntime) crash(path string) { + runtime.mu.Lock() + process := runtime.running[path] + delete(runtime.running, path) + runtime.mu.Unlock() + + process.exit(errors.New("signal: killed")) +} + +func (runtime *fakeRuntime) setConfig(name, content string) { + runtime.mu.Lock() + defer runtime.mu.Unlock() + + if content == "" { + delete(runtime.configs, name) + + return + } + + runtime.configs[name] = []byte(content) +} + +func (runtime *fakeRuntime) snapshot() (starts, unmounts, stops []string, mounted []string) { + runtime.mu.Lock() + defer runtime.mu.Unlock() + + for mountpoint, layers := range runtime.mounted { + if layers > 0 { + mounted = append(mounted, mountpoint) + } + } + + slices.Sort(mounted) + + return slices.Clone(runtime.starts), slices.Clone(runtime.unmounts), slices.Clone(runtime.stopCalls), mounted +} + +// stoppable models mergerfs's SIGTERM behavior: it detaches its own mount. +type stoppable struct { + *fakeProcess + + runtime *fakeRuntime +} + +func (process *stoppable) Stop(context.Context) error { + // Like osProcess.Stop: nothing to do for a process that has exited; its + // leftover mount is the supervisor's problem. + select { + case <-process.done: + return nil + default: + } + + process.runtime.mu.Lock() + process.runtime.stopCalls = append(process.runtime.stopCalls, process.path) + process.runtime.mounted[process.mountpoint]-- + + if process.runtime.running[process.path] == process.fakeProcess { + delete(process.runtime.running, process.path) + } + process.runtime.mu.Unlock() + + process.exit(nil) + + return nil +} + +// mountpointFor resolves the mountpoint of the config at path; tests key +// configs by file name under /etc/mergerfs. +func mountpointFor(configs map[string][]byte, path string) string { + for name, content := range configs { + if "/etc/mergerfs/"+name == path { + return server.Mountpoint(content) + } + } + + return "" +} + +func testConfig() server.Config { + return server.Config{ + Dir: "/etc/mergerfs", + PollInterval: 5 * time.Millisecond, + StopTimeout: 50 * time.Millisecond, + HookWait: 200 * time.Millisecond, + HookPoll: 5 * time.Millisecond, + Logger: log.New(io.Discard, "", 0), + } +} + +type harness struct { + cancel context.CancelFunc + result chan error +} + +func start(t *testing.T, runtime *fakeRuntime) *harness { + t.Helper() + + ctx, cancel := context.WithCancel(t.Context()) + + h := &harness{ + cancel: cancel, + result: make(chan error, 1), + } + + go func() { + h.result <- server.Run(ctx, runtime, testConfig()) + }() + + t.Cleanup(func() { + cancel() + <-h.result + }) + + return h +} + +// wait returns Run's result, leaving it for the cleanup. +func (h *harness) wait(t *testing.T) error { + t.Helper() + + select { + case err := <-h.result: + h.result <- err + + return err + case <-time.After(5 * time.Second): + t.Fatal("Run did not return") + + return nil + } +} + +func eventually(t *testing.T, what string, cond func() bool) { + t.Helper() + + deadline := time.Now().Add(5 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + + time.Sleep(time.Millisecond) + } +} + +const ( + poolConfig = "branches=/var/mnt/ssd:/var/mnt/disk1\nmountpoint=/var/mnt/pool\n" + poolSlowConfig = "# slow tier\nbranches=/var/mnt/disk1\nmountpoint = /var/mnt/pool-slow\n" +) + +func TestStartsOneMergerfsPerConfig(t *testing.T) { + t.Parallel() + + runtime := newFakeRuntime(t, map[string]string{"pool.ini": poolConfig, "pool-slow.ini": poolSlowConfig}) + start(t, runtime) + + eventually(t, "both mounts", func() bool { + _, _, _, mounted := runtime.snapshot() + + return slices.Equal(mounted, []string{"/var/mnt/pool", "/var/mnt/pool-slow"}) + }) + + starts, _, _, _ := runtime.snapshot() + if !slices.Equal(starts, []string{"/etc/mergerfs/pool-slow.ini", "/etc/mergerfs/pool.ini"}) { + t.Fatalf("unexpected starts %v", starts) + } +} + +func TestDetachesLeftoverMountsBeforeStarting(t *testing.T) { + t.Parallel() + + runtime := newFakeRuntime(t, map[string]string{"pool.ini": poolConfig}) + runtime.mounted["/var/mnt/pool"] = 2 // two crashed instances stacked + start(t, runtime) + + // fakeRuntime.Start fails the test if it is asked to mount over a leftover. + eventually(t, "start", func() bool { + starts, _, _, _ := runtime.snapshot() + + return len(starts) == 1 + }) + + _, unmounts, _, mounted := runtime.snapshot() + if !slices.Equal(unmounts, []string{"/var/mnt/pool", "/var/mnt/pool"}) || !slices.Equal(mounted, []string{"/var/mnt/pool"}) { + t.Fatalf("expected the leftovers to be detached first: unmounts %v, mounted %v", unmounts, mounted) + } +} + +func TestExitedMergerfsStopsEverything(t *testing.T) { + t.Parallel() + + runtime := newFakeRuntime(t, map[string]string{"pool.ini": poolConfig, "pool-slow.ini": poolSlowConfig}) + h := start(t, runtime) + + eventually(t, "both mounts", func() bool { + _, _, _, mounted := runtime.snapshot() + + return len(mounted) == 2 + }) + + runtime.crash("/etc/mergerfs/pool.ini") + + err := h.wait(t) + if err == nil || !strings.Contains(err.Error(), "pool.ini: mergerfs exited: signal: killed") { + t.Fatalf("Run() error = %v, want the exited mount named", err) + } + + // The other pool is stopped, the crashed one's leftover detached, and Talos restarts the service. + _, unmounts, stops, mounted := runtime.snapshot() + if !slices.Equal(stops, []string{"/etc/mergerfs/pool-slow.ini"}) || !slices.Equal(unmounts, []string{"/var/mnt/pool"}) || len(mounted) != 0 { + t.Fatalf("stops %v, unmounts %v, mounted %v", stops, unmounts, mounted) + } +} + +func TestConfigChangeStopsEverything(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + change func(*fakeRuntime) + }{ + {"added", func(r *fakeRuntime) { r.setConfig("pool-slow.ini", poolSlowConfig) }}, + {"changed", func(r *fakeRuntime) { r.setConfig("pool.ini", poolConfig+"minfreespace=4G\n") }}, + {"removed", func(r *fakeRuntime) { r.setConfig("pool.ini", "") }}, + } { + runtime := newFakeRuntime(t, map[string]string{"pool.ini": poolConfig}) + h := start(t, runtime) + + eventually(t, "first start", func() bool { + starts, _, _, _ := runtime.snapshot() + + return len(starts) == 1 + }) + + tc.change(runtime) + + if err := h.wait(t); err != nil { + t.Fatalf("%s: Run() error = %v", tc.name, err) + } + + // Nothing is started by this instance: the restarted service reads the new configuration. + starts, _, stops, mounted := runtime.snapshot() + if len(starts) != 1 || !slices.Equal(stops, []string{"/etc/mergerfs/pool.ini"}) || len(mounted) != 0 { + t.Fatalf("%s: starts %v, stops %v, mounted %v", tc.name, starts, stops, mounted) + } + } +} + +func TestRejectsBadConfigs(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + configs map[string]string + want string + }{ + {map[string]string{"broken.ini": "branches=/var/mnt/disk1\n", "pool.ini": poolConfig}, "broken.ini: no mountpoint= in config"}, + {map[string]string{"a.ini": poolConfig, "b.ini": "branches=/var/mnt/other\nmountpoint=/var/mnt/pool\n"}, "b.ini: mountpoint /var/mnt/pool is used by a.ini"}, + } { + runtime := newFakeRuntime(t, tc.configs) + runtime.mounted["/var/mnt/pool"] = 1 // whatever is there is not ours to touch + + err := server.Run(t.Context(), runtime, testConfig()) + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("Run() error = %v, want %q", err, tc.want) + } + + starts, unmounts, _, _ := runtime.snapshot() + if len(starts) != 0 || len(unmounts) != 0 { + t.Fatalf("a rejected configuration touched the host: starts %v, unmounts %v", starts, unmounts) + } + } +} + +func TestDrainStopsEverythingUntilItEnds(t *testing.T) { + t.Parallel() + + runtime := newFakeRuntime(t, map[string]string{"pool.ini": poolConfig, "pool-slow.ini": poolSlowConfig}) + h := start(t, runtime) + + eventually(t, "both mounts", func() bool { + _, _, _, mounted := runtime.snapshot() + + return len(mounted) == 2 + }) + + runtime.setDraining(true) + + if err := h.wait(t); err != nil { + t.Fatalf("Run() error = %v", err) + } + + if _, _, stops, mounted := runtime.snapshot(); len(stops) != 2 || len(mounted) != 0 { + t.Fatalf("expected both mounts stopped: stops %v, mounted %v", stops, mounted) + } + + // The restarted service waits while the request is in force. + start(t, runtime) + time.Sleep(50 * time.Millisecond) + + if starts, _, _, _ := runtime.snapshot(); len(starts) != 2 { + t.Fatalf("mounts came back while draining: starts %v", starts) + } + + // An aborted shutdown: the request expires and everything comes back. + runtime.setDraining(false) + + eventually(t, "both mounts again", func() bool { + _, _, _, mounted := runtime.snapshot() + + return slices.Equal(mounted, []string{"/var/mnt/pool", "/var/mnt/pool-slow"}) + }) +} + +func TestCancelStopsEverything(t *testing.T) { + t.Parallel() + + runtime := newFakeRuntime(t, map[string]string{"pool.ini": poolConfig, "pool-slow.ini": poolSlowConfig}) + h := start(t, runtime) + + eventually(t, "both mounts", func() bool { + _, _, _, mounted := runtime.snapshot() + + return len(mounted) == 2 + }) + + h.cancel() + + if err := h.wait(t); err != nil { + t.Fatalf("Run() error = %v", err) + } + + _, _, stops, mounted := runtime.snapshot() + if len(stops) != 2 || len(mounted) != 0 { + t.Fatalf("expected both mounts stopped, got stops %v, mounted %v", stops, mounted) + } +} + +func TestMountpoint(t *testing.T) { + t.Parallel() + + for name, tc := range map[string]struct { + config string + want string + }{ + "plain": {"branches=/a:/b\nmountpoint=/var/mnt/pool\n", "/var/mnt/pool"}, + "spaces": {" mountpoint = /var/mnt/pool \n", "/var/mnt/pool"}, + "trailing slash": {"mountpoint=/var/mnt/pool/\n", "/var/mnt/pool"}, + "alias": {"mount=/var/mnt/pool\n", "/var/mnt/pool"}, + "last wins": {"mountpoint=/var/mnt/a\nmountpoint=/var/mnt/b\n", "/var/mnt/b"}, + "comment ignored": {"# mountpoint=/var/mnt/a\nmountpoint=/var/mnt/b\n", "/var/mnt/b"}, + "no newline": {"mountpoint=/var/mnt/pool", "/var/mnt/pool"}, + "missing": {"branches=/a\n", ""}, + "flag without =": {"mountpoint\n", ""}, + } { + if got := server.Mountpoint([]byte(tc.config)); got != tc.want { + t.Errorf("%s: Mountpoint() = %q, want %q", name, got, tc.want) + } + } +} + +func TestManifestRequiresHostRunnerCapableTalos(t *testing.T) { + t.Parallel() + + contents, err := os.ReadFile("../../manifest.yaml.tmpl") + if err != nil { + t.Fatalf("read manifest: %v", err) + } + + if !slices.Contains([]string{">= v1.14.0", ">= v1.15.0"}, manifestTalosVersion(string(contents))) { + t.Fatalf("manifest Talos compatibility does not require host-runner support: %s", contents) + } +} + +func manifestTalosVersion(contents string) string { + const prefix = " version: \"" + + for _, line := range strings.Split(contents, "\n") { + if strings.HasPrefix(line, prefix) && strings.HasSuffix(line, "\"") { + return strings.TrimSuffix(strings.TrimPrefix(line, prefix), "\"") + } + } + + return "" +} diff --git a/storage/mergerfs/pkg.yaml b/storage/mergerfs/pkg.yaml new file mode 100644 index 00000000..9600f669 --- /dev/null +++ b/storage/mergerfs/pkg.yaml @@ -0,0 +1,64 @@ +name: mergerfs +variant: scratch +shell: /bin/bash +dependencies: + - stage: base +steps: + - sources: + - url: https://github.com/trapexit/mergerfs/releases/download/{{ .MERGERFS_VERSION }}/mergerfs-{{ .MERGERFS_VERSION }}.tar.gz + destination: mergerfs.tar.gz + sha256: {{ .MERGERFS_SHA256 }} + sha512: {{ .MERGERFS_SHA512 }} + env: + SOURCE_DATE_EPOCH: {{ .BUILD_ARG_SOURCE_DATE_EPOCH }} + prepare: + - | + tar -xzf mergerfs.tar.gz --strip-components=1 + build: + - | + # The Makefile only applies its own flags (including -static, -flto + # and -DNDEBUG) when CFLAGS/CXXFLAGS are unset, so drop the bldr + # defaults. Same flags as upstream's static releases; libfuse is vendored. + unset CFLAGS CXXFLAGS LDFLAGS + + make NDEBUG=1 LTO=1 STATIC=1 LDFLAGS=-s -j $(nproc) build/mergerfs + - | + cd /pkg/mounts + + go test ./... + CGO_ENABLED=0 go build -ldflags "-s -w" -trimpath -o /tmp/mergerfs-supervisor ./cmd/mergerfs-supervisor + install: + - | + mkdir -p /rootfs/usr/local/bin /rootfs/usr/local/libexec /rootfs/usr/local/etc/containers + + cp build/mergerfs /rootfs/usr/local/bin/mergerfs + cp /tmp/mergerfs-supervisor /rootfs/usr/local/libexec/mergerfs-supervisor + cp /pkg/mergerfs.yaml /rootfs/usr/local/etc/containers/ + test: + - | + mkdir -p /extensions-validator-rootfs + cp -r /rootfs/ /extensions-validator-rootfs/rootfs + cp /pkg/manifest.yaml /extensions-validator-rootfs/manifest.yaml + /extensions-validator validate --rootfs=/extensions-validator-rootfs --pkg-name="${PKG_NAME}" + - | + [[ $(/rootfs/usr/local/bin/mergerfs --version) == *{{ .MERGERFS_VERSION }}* ]] + - | + # The host service runs it straight from the Talos rootfs: no loader, no shared libraries. + readelf -l /rootfs/usr/local/bin/mergerfs > /tmp/mergerfs.phdrs + grep -q ' LOAD ' /tmp/mergerfs.phdrs + if grep -q INTERP /tmp/mergerfs.phdrs; then + echo "mergerfs is dynamically linked" >&2 + exit 1 + fi + sbom: + outputPath: /rootfs/usr/local/share/spdx/mergerfs.spdx.json + version: {{ .MERGERFS_VERSION }} + # mergerfs, and the libfuse fork it vendors and links statically. + licenses: + - ISC + - LGPL-2.1 +finalize: + - from: /rootfs + to: /rootfs + - from: /pkg/manifest.yaml + to: / diff --git a/storage/mergerfs/vars.yaml b/storage/mergerfs/vars.yaml new file mode 100644 index 00000000..41186f65 --- /dev/null +++ b/storage/mergerfs/vars.yaml @@ -0,0 +1,2 @@ +VERSION: "{{ .MERGERFS_VERSION }}" +TIER: "contrib" diff --git a/storage/vars.yaml b/storage/vars.yaml index 68c303f2..f53a24f2 100644 --- a/storage/vars.yaml +++ b/storage/vars.yaml @@ -18,3 +18,7 @@ SQLITE_SHA512: 5975efe014f43b974b1dd88be36b3a37ae97ec11eec1b9159a853cd6eb2898035 CACHEFILESD_VERSION: "0.10.10" CACHEFILESD_SHA256: 0d0309851efabd02b7c849f73535b8ad3f831570e83e4f65e42354da18e11a02 CACHEFILESD_SHA512: d7d816b5ef1fffe1272cb8c2e9cbd18c1393438afca250436a36a446c6a37303e7784057725a56be839e0489101190b563c4fc015fc4ff11baa8003121e5183a +# renovate: datasource=github-releases depName=trapexit/mergerfs +MERGERFS_VERSION: 2.42.0 +MERGERFS_SHA256: 546d6eed694d48ef13e058a46c6fb3da34dc9dfc48ee51af6487f353787780bd +MERGERFS_SHA512: 90fdbb006bf8d354e872ccbd864b911f21865cd20fdeb65f1694a04b4d108810fe4eabe8e4684f0d4471aecbf854a5ab302fa6e87b1b7f63276c28376bde5b43