diff --git a/agent/agent.go b/agent/agent.go index 0dd8638e..b3598f4e 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -48,6 +48,7 @@ type Agent struct { keys []gossh.PublicKey // SSH public keys smartManager *SmartManager // Manages SMART data systemdManager *systemdManager // Manages systemd services + zfsManager *ZfsManager // Manages ZFS pool and dataset data } // NewAgent creates a new agent with the given data directory for persisting data. @@ -121,6 +122,19 @@ func NewAgent(dataDir ...string) (agent *Agent, err error) { // initialize handler registry agent.handlerRegistry = NewHandlerRegistry() + agent.zfsManager = newZfsManager() + + // ZFS_INTERVAL env var to update ZFS detail data at this interval + if zfsIntervalEnv, exists := utils.GetEnv("ZFS_INTERVAL"); exists { + if duration, err := time.ParseDuration(zfsIntervalEnv); err == nil && duration > 0 { + agent.zfsManager.detailInterval = duration + agent.systemDetails.ZfsInterval = duration + slog.Info("ZFS_INTERVAL", "duration", duration) + } else { + slog.Warn("Invalid ZFS_INTERVAL", "err", err) + } + } + // initialize disk info agent.initializeDiskInfo() diff --git a/agent/disk.go b/agent/disk.go index e76a999e..fa96fb23 100644 --- a/agent/disk.go +++ b/agent/disk.go @@ -537,7 +537,16 @@ func normalizeDeviceName(value string) string { func (a *Agent) initializeDiskIoStats(diskIoCounters map[string]disk.IOCountersStat) { a.fsNames = a.fsNames[:0] now := time.Now() + // ZFS datasets have no /proc/diskstats entry, so they are excluded from + // I/O tracking instead of warning about a missing device (#1541). + var zfsMountpoints map[string]bool + if a.zfsManager != nil { + zfsMountpoints = a.zfsManager.ZfsMountpoints() + } for device, stats := range a.fsStats { + if zfsMountpoints[stats.Mountpoint] { + continue + } // skip if not in diskIoCounters d, exists := diskIoCounters[device] if !exists { @@ -562,20 +571,31 @@ func (a *Agent) updateDiskUsage(systemStats *system.Stats) { !a.lastDiskUsageUpdate.IsZero() && time.Since(a.lastDiskUsageUpdate) < a.diskUsageCacheDuration + // ZFS dataset mountpoints use `zfs list` values because statfs(2) reports + // dataset-level usage that excludes child datasets (#1541). + var zfsUsage map[string]zfsDatasetUsage + if a.zfsManager != nil { + zfsUsage = a.zfsManager.DatasetUsage() + } + // disk usage for _, stats := range a.fsStats { // Skip non-root filesystems if caching is active if cacheExtraFs && !stats.Root { continue } - if d, err := disk.Usage(stats.Mountpoint); err == nil { - stats.DiskTotal = utils.BytesToGigabytes(d.Total) - stats.DiskUsed = utils.BytesToGigabytes(d.Used) - if stats.Root { - systemStats.DiskTotal = utils.BytesToGigabytes(d.Total) - systemStats.DiskUsed = utils.BytesToGigabytes(d.Used) - systemStats.DiskPct = utils.TwoDecimals(d.UsedPercent) + var total, used uint64 + var usedPct float64 + if u, ok := zfsUsage[stats.Mountpoint]; ok { + total = u.used + u.avail + used = u.used + if total > 0 { + usedPct = float64(used) / float64(total) * 100 } + } else if d, err := disk.Usage(stats.Mountpoint); err == nil { + total = d.Total + used = d.Used + usedPct = d.UsedPercent } else { // reset stats if error (likely unmounted) slog.Error("Error getting disk stats", "name", stats.Mountpoint, "err", err) @@ -583,6 +603,14 @@ func (a *Agent) updateDiskUsage(systemStats *system.Stats) { stats.DiskUsed = 0 stats.TotalRead = 0 stats.TotalWrite = 0 + continue + } + stats.DiskTotal = utils.BytesToGigabytes(total) + stats.DiskUsed = utils.BytesToGigabytes(used) + if stats.Root { + systemStats.DiskTotal = stats.DiskTotal + systemStats.DiskUsed = stats.DiskUsed + systemStats.DiskPct = utils.TwoDecimals(usedPct) } } diff --git a/agent/disk_zfs_test.go b/agent/disk_zfs_test.go new file mode 100644 index 00000000..5f0f3f82 --- /dev/null +++ b/agent/disk_zfs_test.go @@ -0,0 +1,109 @@ +//go:build testing + +package agent + +import ( + "testing" + + "github.com/henrygd/beszel/agent/zfs" + "github.com/henrygd/beszel/internal/entities/system" + "github.com/shirou/gopsutil/v4/disk" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestUpdateDiskUsageZfsMountpoint verifies that a filesystem whose mountpoint +// is a ZFS dataset reports `zfs list` usage (which includes child datasets) +// instead of the dataset-scoped statfs values (#1541). +func TestUpdateDiskUsageZfsMountpoint(t *testing.T) { + zm := &ZfsManager{} + zm.datasetsFn = func() ([]zfs.Dataset, error) { + return []zfs.Dataset{ + {Name: "tank", Used: 12000000000000, Avail: 11999000000000, Mountpoint: "/tank"}, + }, nil + } + agent := &Agent{ + fsStats: map[string]*system.FsStats{ + "tank": {Root: false, Mountpoint: "/tank"}, + }, + zfsManager: zm, + } + + var stats system.Stats + agent.updateDiskUsage(&stats) + + fs := agent.fsStats["tank"] + require.NotNil(t, fs) + assert.Equal(t, 22350.81, fs.DiskTotal) // (used + avail) in GiB + assert.Equal(t, 11175.87, fs.DiskUsed) + // Non-root filesystems do not populate system-level stats. + assert.Equal(t, float64(0), stats.DiskTotal) +} + +// TestUpdateDiskUsageZfsRootPopulatesSystemStats verifies the root disk values +// are derived from ZFS usage when the root mountpoint is a ZFS dataset. +func TestUpdateDiskUsageZfsRootPopulatesSystemStats(t *testing.T) { + zm := &ZfsManager{} + zm.datasetsFn = func() ([]zfs.Dataset, error) { + return []zfs.Dataset{ + {Name: "rpool/ROOT/pve-1", Used: 900000000000, Avail: 300000000000, Mountpoint: "/"}, + }, nil + } + agent := &Agent{ + fsStats: map[string]*system.FsStats{ + "rpool/ROOT/pve-1": {Root: true, Mountpoint: "/"}, + }, + zfsManager: zm, + } + + var stats system.Stats + agent.updateDiskUsage(&stats) + + assert.Equal(t, 1117.59, agent.fsStats["rpool/ROOT/pve-1"].DiskTotal) + assert.Equal(t, 838.19, agent.fsStats["rpool/ROOT/pve-1"].DiskUsed) + assert.Equal(t, 75.0, stats.DiskPct) + assert.Equal(t, 1117.59, stats.DiskTotal) + assert.Equal(t, 838.19, stats.DiskUsed) +} + +// TestUpdateDiskUsageWithoutZfsManager falls back to statfs when no manager is +// present (e.g. tests constructing bare Agent values). +func TestUpdateDiskUsageWithoutZfsManager(t *testing.T) { + agent := &Agent{ + fsStats: map[string]*system.FsStats{ + "root": {Root: true, Mountpoint: "/"}, + }, + } + + var stats system.Stats + agent.updateDiskUsage(&stats) + + assert.True(t, agent.fsStats["root"].DiskTotal > 0, "root usage should come from statfs") + assert.True(t, stats.DiskTotal > 0) +} + +// TestInitializeDiskIoStatsSkipsZfsMountpoints verifies ZFS filesystems are +// excluded from diskstats I/O tracking instead of warning about a missing device. +func TestInitializeDiskIoStatsSkipsZfsMountpoints(t *testing.T) { + zm := &ZfsManager{} + zm.datasetsFn = func() ([]zfs.Dataset, error) { + return []zfs.Dataset{{Name: "tank", Mountpoint: "/tank"}}, nil + } + agent := &Agent{ + fsStats: map[string]*system.FsStats{ + "tank": {Root: false, Mountpoint: "/tank"}, + "sda1": {Root: false, Mountpoint: "/mnt/data"}, + }, + zfsManager: zm, + diskPrev: make(map[uint16]map[string]prevDisk), + } + + agent.initializeDiskIoStats(map[string]disk.IOCountersStat{ + "sda1": {Name: "sda1", ReadBytes: 100, WriteBytes: 100}, + }) + + assert.Equal(t, []string{"sda1"}, agent.fsNames) + assert.Equal(t, uint64(100), agent.fsStats["sda1"].TotalRead) + // ZFS entry is present but untouched by diskstats initialization. + assert.Equal(t, uint64(0), agent.fsStats["tank"].TotalRead) +} diff --git a/agent/handlers.go b/agent/handlers.go index adf9c918..4c8eb30e 100644 --- a/agent/handlers.go +++ b/agent/handlers.go @@ -51,6 +51,7 @@ func NewHandlerRegistry() *HandlerRegistry { registry.Register(common.GetContainerInfo, &GetContainerInfoHandler{}) registry.Register(common.GetSmartData, &GetSmartDataHandler{}) registry.Register(common.GetSystemdInfo, &GetSystemdInfoHandler{}) + registry.Register(common.GetZfsData, &GetZfsDataHandler{}) return registry } @@ -178,6 +179,23 @@ func (h *GetSmartDataHandler) Handle(hctx *HandlerContext) error { }, hctx.RequestID) } +//////////////////////////////////////////////////////////////////////////// +//////////////////////////////////////////////////////////////////////////// + +// GetZfsDataHandler handles ZFS detail data requests +type GetZfsDataHandler struct{} + +func (h *GetZfsDataHandler) Handle(hctx *HandlerContext) error { + if hctx.Agent.zfsManager == nil { + return hctx.SendResponse(nil, hctx.RequestID) + } + var req common.ZfsDataRequest + if err := cbor.Unmarshal(hctx.Request.Data, &req); err != nil { + return err + } + return hctx.SendResponse(hctx.Agent.zfsManager.GetDetail(req.Force), hctx.RequestID) +} + //////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////// diff --git a/agent/handlers_test.go b/agent/handlers_test.go index df7fa869..3150d5b7 100644 --- a/agent/handlers_test.go +++ b/agent/handlers_test.go @@ -4,8 +4,10 @@ package agent import ( "testing" + "time" "github.com/fxamacker/cbor/v2" + "github.com/henrygd/beszel/agent/zfs" "github.com/henrygd/beszel/internal/common" "github.com/henrygd/beszel/internal/entities/smart" "github.com/stretchr/testify/assert" @@ -30,6 +32,32 @@ func TestNewAgentResponseSmartData(t *testing.T) { assert.True(t, response.SmartComplete) } +func TestGetZfsDataHandlerForceRefresh(t *testing.T) { + poolCalls := 0 + zm := &ZfsManager{detailInterval: time.Hour} + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { + poolCalls++ + return []zfs.PoolStat{{Name: "tank", Alloc: uint64(poolCalls)}}, nil + } + zm.poolStatusesFn = func() ([]zfs.PoolStatus, error) { return nil, nil } + zm.datasetsFn = func() ([]zfs.Dataset, error) { return nil, nil } + zm.GetDetail(false) + + requestData, err := cbor.Marshal(common.ZfsDataRequest{Force: true}) + assert.NoError(t, err) + ctx := &HandlerContext{ + Agent: &Agent{zfsManager: zm}, + Request: &common.HubRequest[cbor.RawMessage]{ + Action: common.GetZfsData, + Data: requestData, + }, + SendResponse: func(any, *uint32) error { return nil }, + } + + assert.NoError(t, (&GetZfsDataHandler{}).Handle(ctx)) + assert.Equal(t, 2, poolCalls) +} + func (m *MockHandler) Handle(ctx *HandlerContext) error { if m.handleFunc != nil { return m.handleFunc(ctx) diff --git a/agent/system.go b/agent/system.go index 1df167ee..ff322958 100644 --- a/agent/system.go +++ b/agent/system.go @@ -219,6 +219,9 @@ func (a *Agent) getSystemStats(cacheTimeMs uint16) system.Stats { // disk i/o (cache-aware per interval) a.updateDiskIo(cacheTimeMs, &systemStats) + // zfs pool stats + a.zfsManager.Update(&systemStats) + // network stats (per cache interval) a.updateNetworkStats(cacheTimeMs, &systemStats) diff --git a/agent/test-data/zfs/zfs_list.txt b/agent/test-data/zfs/zfs_list.txt new file mode 100644 index 00000000..68a020df --- /dev/null +++ b/agent/test-data/zfs/zfs_list.txt @@ -0,0 +1,9 @@ +tank 12000000000000 11999000000000 /tank +tank/apps 1000000000000 11999000000000 /tank/apps +tank/backup 2000000000000 11999000000000 /tank/backup +tank/media 1000000000000 11999000000000 /tank/my media +rpool 900000000000 300000000000 - +rpool/ROOT 1000000000 300000000000 - +rpool/ROOT/pve-1 890000000000 300000000000 / +rpool/data 9000000000 300000000000 - +rpool/data/subvol-100-disk-0 400000000000 300000000000 /subvol-100-disk-0 diff --git a/agent/test-data/zfs/zpool_list.txt b/agent/test-data/zfs/zpool_list.txt new file mode 100644 index 00000000..9938bb90 --- /dev/null +++ b/agent/test-data/zfs/zpool_list.txt @@ -0,0 +1,2 @@ +tank 23999000000000 12000000000000 11999000000000 ONLINE +rpool 1200000000000 900000000000 300000000000 DEGRADED diff --git a/agent/test-data/zfs/zpool_status.txt b/agent/test-data/zfs/zpool_status.txt new file mode 100644 index 00000000..e25076c8 --- /dev/null +++ b/agent/test-data/zfs/zpool_status.txt @@ -0,0 +1,29 @@ + pool: tank + state: ONLINE + scan: scrub repaired 0B in 00:05:12 with 0 errors on Sun Jun 1 02:00:12 2025 +config: + + NAME STATE READ WRITE CKSUM + tank ONLINE 0 0 0 + mirror-0 ONLINE 0 0 0 + sda ONLINE 0 0 0 + sdb ONLINE 0 0 0 + +errors: No known data errors + + pool: rpool + state: DEGRADED + status: One or more devices could not be used because the label is missing or + invalid. Sufficient replicas exist for the pool to continue functioning in a + degraded state. + scan: scrub in progress since Sun Jun 8 01:00:00 2025 + 10.00% done, 01:30:00 to go, 0.00/s +config: + + NAME STATE READ WRITE CKSUM + rpool DEGRADED 0 0 0 + mirror-0 DEGRADED 0 0 0 + sda ONLINE 0 0 0 + sdb FAULTED 1 2 3 + +errors: 1 data errors, use '-v' for a list diff --git a/agent/zfs/zfs.go b/agent/zfs/zfs.go new file mode 100644 index 00000000..1f8f2258 --- /dev/null +++ b/agent/zfs/zfs.go @@ -0,0 +1,160 @@ +// Package zfs provides functions to read ZFS statistics. +package zfs + +import ( + "bufio" + "bytes" + "context" + "errors" + "fmt" + "os" + "os/exec" + "strconv" + "strings" + "time" +) + +var commandTimeout = 10 * time.Second + +var commandOutput = func(name string, args ...string) ([]byte, error) { + ctx, cancel := context.WithTimeout(context.Background(), commandTimeout) + defer cancel() + cmd := exec.CommandContext(ctx, name, args...) + cmd.Env = append(os.Environ(), "LC_ALL=C", "LANG=C") + out, err := cmd.Output() + if ctx.Err() != nil { + return nil, fmt.Errorf("%s timed out after %s: %w", name, commandTimeout, ctx.Err()) + } + return out, err +} + +// ErrNoZfs is returned when the ZFS utilities or kernel interfaces are unavailable. +var ErrNoZfs = errors.New("zfs utilities unavailable") + +// PoolStat is a snapshot of a ZFS pool's capacity and health. +type PoolStat struct { + Name string + Size uint64 // total capacity in bytes + Alloc uint64 // allocated bytes + Free uint64 // free bytes + Health string // ONLINE, DEGRADED, FAULTED, ... +} + +// PoolKernelStat is the inexpensive pool telemetry exposed by the ZFS kernel. +// NRead and NWrite are cumulative byte counters since the pool was imported. +type PoolKernelStat struct { + Name string + Health string + NRead uint64 + NWrite uint64 +} + +// PoolIoStats holds calculated per-second I/O rates for a pool. +type PoolIoStats struct { + NRead uint64 + NWrite uint64 +} + +// Dataset is a single ZFS dataset with usage information. +type Dataset struct { + Name string + Used uint64 + Avail uint64 + Mountpoint string +} + +// PoolStats returns capacity and health for all pools on the system using +// `zpool list`. Frequent health and I/O sampling uses PoolKernelStats instead. +func PoolStats() ([]PoolStat, error) { + out, err := commandOutput("zpool", "list", "-Hp", "-o", "name,size,alloc,free,health") + if err != nil { + var exitErr *exec.ExitError + if errors.As(err, &exitErr) && strings.Contains(string(exitErr.Stderr), "no pools available") { + return nil, nil + } + return nil, fmt.Errorf("zpool list: %w", err) + } + return parseZpoolListOutput(out) +} + +// Datasets returns all datasets on the system with usage and mountpoint +// information using `zfs list` (recursive by default). +func Datasets() ([]Dataset, error) { + out, err := commandOutput("zfs", "list", "-Hp", "-o", "name,used,avail,mountpoint") + if err != nil { + return nil, fmt.Errorf("zfs list: %w", err) + } + return parseZfsListOutput(out) +} + +// parseZpoolListOutput parses `zpool list -Hp -o name,size,alloc,free,health` output. +// Columns are tab-separated; numeric columns are raw bytes. +func parseZpoolListOutput(out []byte) ([]PoolStat, error) { + var pools []PoolStat + scanner := bufio.NewScanner(bytes.NewReader(out)) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" { + continue + } + if line == "no pools available" && len(pools) == 0 { + return nil, nil + } + fields := strings.Split(line, "\t") + if len(fields) < 5 { + return nil, fmt.Errorf("unexpected zpool list line: %q", line) + } + size, err := strconv.ParseUint(fields[1], 10, 64) + if err != nil { + return nil, fmt.Errorf("parsing size for pool %q: %w", fields[0], err) + } + alloc, err := strconv.ParseUint(fields[2], 10, 64) + if err != nil { + return nil, fmt.Errorf("parsing alloc for pool %q: %w", fields[0], err) + } + free, err := strconv.ParseUint(fields[3], 10, 64) + if err != nil { + return nil, fmt.Errorf("parsing free for pool %q: %w", fields[0], err) + } + pools = append(pools, PoolStat{ + Name: fields[0], + Size: size, + Alloc: alloc, + Free: free, + Health: fields[4], + }) + } + return pools, scanner.Err() +} + +// parseZfsListOutput parses `zfs list -Hp -o name,used,avail,mountpoint` output. +// The mountpoint column may contain spaces, so it is split on tabs only. +func parseZfsListOutput(out []byte) ([]Dataset, error) { + var datasets []Dataset + scanner := bufio.NewScanner(bytes.NewReader(out)) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" { + continue + } + fields := strings.SplitN(line, "\t", 4) + if len(fields) < 4 { + return nil, fmt.Errorf("unexpected zfs list line: %q", line) + } + used, err := strconv.ParseUint(fields[1], 10, 64) + if err != nil { + return nil, fmt.Errorf("parsing used for dataset %q: %w", fields[0], err) + } + avail, err := strconv.ParseUint(fields[2], 10, 64) + if err != nil { + return nil, fmt.Errorf("parsing avail for dataset %q: %w", fields[0], err) + } + datasets = append(datasets, Dataset{ + Name: fields[0], + Used: used, + Avail: avail, + Mountpoint: fields[3], + }) + } + return datasets, scanner.Err() +} diff --git a/agent/zfs/zfs_freebsd.go b/agent/zfs/zfs_freebsd.go index 869776a3..e0bcd04e 100644 --- a/agent/zfs/zfs_freebsd.go +++ b/agent/zfs/zfs_freebsd.go @@ -3,9 +3,17 @@ package zfs import ( + "errors" + "golang.org/x/sys/unix" ) func ARCSize() (uint64, error) { return unix.SysctlUint64("kstat.zfs.misc.arcstats.size") } + +// FreeBSD does not expose Linux's per-pool procfs kstats. Capacity, health, +// and detail collection still work through the cached utilities. +func PoolKernelStats() ([]PoolKernelStat, error) { + return nil, errors.ErrUnsupported +} diff --git a/agent/zfs/zfs_linux.go b/agent/zfs/zfs_linux.go index 9f94380d..6be9b90e 100644 --- a/agent/zfs/zfs_linux.go +++ b/agent/zfs/zfs_linux.go @@ -5,14 +5,18 @@ package zfs import ( "bufio" + "errors" "fmt" "os" + "path/filepath" "strconv" "strings" ) +var procZfsPath = "/proc/spl/kstat/zfs" + func ARCSize() (uint64, error) { - file, err := os.Open("/proc/spl/kstat/zfs/arcstats") + file, err := os.Open(filepath.Join(procZfsPath, "arcstats")) if err != nil { return 0, err } @@ -29,6 +33,167 @@ func ARCSize() (uint64, error) { return strconv.ParseUint(fields[2], 10, 64) } } + if err := scanner.Err(); err != nil { + return 0, err + } return 0, fmt.Errorf("size field not found in arcstats") } + +// PoolKernelStats reads pool state and cumulative I/O counters directly from +// procfs. These kstats are the same interfaces used by node_exporter's Linux +// ZFS collector and avoid keeping a `zpool iostat` subprocess alive. +func PoolKernelStats() ([]PoolKernelStat, error) { + poolDirs := make(map[string]struct{}) + for _, filename := range []string{"state", "io", "objset-*"} { + paths, err := filepath.Glob(filepath.Join(procZfsPath, "*", filename)) + if err != nil { + return nil, err + } + for _, path := range paths { + poolDirs[filepath.Dir(path)] = struct{}{} + } + } + if len(poolDirs) == 0 { + return nil, ErrNoZfs + } + pools := make([]PoolKernelStat, 0, len(poolDirs)) + for poolDir := range poolDirs { + nread, nwrite, err := readPoolCounters(poolDir) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + continue // pool may have been exported after the glob + } + return nil, err + } + state, err := os.ReadFile(filepath.Join(poolDir, "state")) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return nil, err + } + pools = append(pools, PoolKernelStat{ + Name: filepath.Base(poolDir), Health: strings.ToUpper(strings.TrimSpace(string(state))), + NRead: nread, NWrite: nwrite, + }) + } + if len(pools) == 0 { + return nil, ErrNoZfs + } + return pools, nil +} + +// readPoolCounters supports both ZFS kernel interfaces. OpenZFS through 2.3 +// exposes aggregate vdev counters in "io". When that file is unavailable, sum +// the logical I/O counters exposed for each dataset in the pool. +func readPoolCounters(poolDir string) (uint64, uint64, error) { + nread, nwrite, err := readPoolIO(filepath.Join(poolDir, "io")) + if err == nil || !errors.Is(err, os.ErrNotExist) { + return nread, nwrite, err + } + return readPoolObjsets(poolDir) +} + +func readPoolIO(path string) (uint64, uint64, error) { + file, err := os.Open(path) + if err != nil { + return 0, 0, err + } + defer file.Close() + + scanner := bufio.NewScanner(file) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) < 2 || fields[0] != "nread" { + continue + } + if !scanner.Scan() { + break + } + values := strings.Fields(scanner.Text()) + if len(values) < 2 { + break + } + nread, err := strconv.ParseUint(values[0], 10, 64) + if err != nil { + return 0, 0, fmt.Errorf("parsing nread in %s: %w", path, err) + } + nwrite, err := strconv.ParseUint(values[1], 10, 64) + if err != nil { + return 0, 0, fmt.Errorf("parsing nwritten in %s: %w", path, err) + } + return nread, nwrite, nil + } + if err := scanner.Err(); err != nil { + return 0, 0, err + } + return 0, 0, fmt.Errorf("I/O counters not found in %s", path) +} + +func readPoolObjsets(poolDir string) (uint64, uint64, error) { + paths, err := filepath.Glob(filepath.Join(poolDir, "objset-*")) + if err != nil { + return 0, 0, err + } + if len(paths) == 0 { + return 0, 0, fmt.Errorf("dataset I/O counters not found in %s", poolDir) + } + + var totalRead, totalWrite uint64 + objsetsRead := 0 + for _, path := range paths { + nread, nwrite, err := readObjsetIO(path) + if errors.Is(err, os.ErrNotExist) { + continue // dataset may have been destroyed after the glob + } + if err != nil { + return 0, 0, err + } + totalRead += nread + totalWrite += nwrite + objsetsRead++ + } + if objsetsRead == 0 { + return 0, 0, fmt.Errorf("dataset I/O counters not found in %s", poolDir) + } + return totalRead, totalWrite, nil +} + +func readObjsetIO(path string) (uint64, uint64, error) { + file, err := os.Open(path) + if err != nil { + return 0, 0, err + } + defer file.Close() + + var nread, nwrite uint64 + var foundRead, foundWrite bool + scanner := bufio.NewScanner(file) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) < 3 { + continue + } + var target *uint64 + switch fields[0] { + case "nread": + target = &nread + foundRead = true + case "nwritten": + target = &nwrite + foundWrite = true + default: + continue + } + value, err := strconv.ParseUint(fields[2], 10, 64) + if err != nil { + return 0, 0, fmt.Errorf("parsing %s in %s: %w", fields[0], path, err) + } + *target = value + } + if err := scanner.Err(); err != nil { + return 0, 0, err + } + if !foundRead || !foundWrite { + return 0, 0, fmt.Errorf("incomplete I/O counters in %s", path) + } + return nread, nwrite, nil +} diff --git a/agent/zfs/zfs_linux_test.go b/agent/zfs/zfs_linux_test.go new file mode 100644 index 00000000..8c8e8623 --- /dev/null +++ b/agent/zfs/zfs_linux_test.go @@ -0,0 +1,90 @@ +//go:build testing && linux + +package zfs + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestPoolKernelStats(t *testing.T) { + root := t.TempDir() + oldPath := procZfsPath + procZfsPath = root + t.Cleanup(func() { procZfsPath = oldPath }) + + poolDir := filepath.Join(root, "tank") + require.NoError(t, os.MkdirAll(poolDir, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(poolDir, "io"), []byte( + "11 3 0x00 1 80 0 0\n"+ + "nread nwritten reads writes wtime wlentime wupdate rtime rlentime rupdate wcnt rcnt\n"+ + "1884160 6450688 22 978 0 0 0 0 0 0 0 0\n", + ), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(poolDir, "state"), []byte("DEGRADED\n"), 0o644)) + + stats, err := PoolKernelStats() + require.NoError(t, err) + require.Len(t, stats, 1) + assert.Equal(t, PoolKernelStat{ + Name: "tank", Health: "DEGRADED", NRead: 1884160, NWrite: 6450688, + }, stats[0]) +} + +func TestPoolKernelStatsOpenZfs24(t *testing.T) { + root := t.TempDir() + oldPath := procZfsPath + procZfsPath = root + t.Cleanup(func() { procZfsPath = oldPath }) + + poolDir := filepath.Join(root, "tank") + require.NoError(t, os.MkdirAll(poolDir, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(poolDir, "state"), []byte("ONLINE\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(poolDir, "objset-0x1"), []byte( + "34 1 0x01 28 7872 0 0\n"+ + "name type data\n"+ + "dataset_name 7 tank\n"+ + "nwritten 4 2000\n"+ + "nread 4 1000\n", + ), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(poolDir, "objset-0x2"), []byte( + "34 1 0x01 28 7872 0 0\n"+ + "name type data\n"+ + "dataset_name 7 tank/videos\n"+ + "nwritten 4 400\n"+ + "nread 4 300\n", + ), 0o644)) + + stats, err := PoolKernelStats() + require.NoError(t, err) + require.Len(t, stats, 1) + assert.Equal(t, PoolKernelStat{ + Name: "tank", Health: "ONLINE", NRead: 1300, NWrite: 2400, + }, stats[0]) +} + +func TestPoolKernelStatsNoZfs(t *testing.T) { + oldPath := procZfsPath + procZfsPath = t.TempDir() + t.Cleanup(func() { procZfsPath = oldPath }) + + _, err := PoolKernelStats() + assert.ErrorIs(t, err, ErrNoZfs) +} + +func TestReadPoolIORejectsMalformedCounters(t *testing.T) { + path := filepath.Join(t.TempDir(), "io") + require.NoError(t, os.WriteFile(path, []byte("nread nwritten\nnope 10\n"), 0o644)) + _, _, err := readPoolIO(path) + require.Error(t, err) +} + +func TestReadObjsetIORequiresAllCounters(t *testing.T) { + path := filepath.Join(t.TempDir(), "objset-0x1") + require.NoError(t, os.WriteFile(path, []byte("nread 4 10\n"), 0o644)) + _, _, err := readObjsetIO(path) + require.Error(t, err) +} diff --git a/agent/zfs/zfs_status.go b/agent/zfs/zfs_status.go new file mode 100644 index 00000000..052d7e72 --- /dev/null +++ b/agent/zfs/zfs_status.go @@ -0,0 +1,150 @@ +package zfs + +import ( + "bufio" + "bytes" + "fmt" + "regexp" + "strconv" + "strings" +) + +// PoolStatus holds parsed `zpool status` information for one pool. +type PoolStatus struct { + Name string + State string // ONLINE, DEGRADED, FAULTED, ... + Scrub ScrubStatus + Vdevs []VdevStatus +} + +// ScrubStatus holds the scrub (or resilver) status parsed from the scan line. +type ScrubStatus struct { + State string // NONE, SCANNING, FINISHED, CANCELED + Progress string // e.g. "10.00%" while scanning + Errors uint64 +} + +// VdevStatus is a single vdev row (mirror, raidz, or leaf disk). +type VdevStatus struct { + Name string + State string + ReadErrs uint64 + WriteErrs uint64 + ChecksumErrs uint64 +} + +var ( + progressRe = regexp.MustCompile(`(\d+\.\d+)%\s+done`) + errorsRe = regexp.MustCompile(`with\s+(\d+)\s+errors`) +) + +// PoolStatuses runs `zpool status` and parses per-pool state, scrub, and vdev +// information. The human-readable format has been stable across OpenZFS +// releases; rows are matched by their tabular shape rather than position. +func PoolStatuses() ([]PoolStatus, error) { + out, err := commandOutput("zpool", "status") + if err != nil { + return nil, fmt.Errorf("zpool status: %w", err) + } + return parseZpoolStatusOutput(out) +} + +// parseZpoolStatusOutput parses the output of `zpool status`. +func parseZpoolStatusOutput(out []byte) ([]PoolStatus, error) { + var pools []PoolStatus + var current *PoolStatus + inConfig := false + scanContinuation := false // next non-blank line continues the scan line (progress) + + scanner := bufio.NewScanner(bytes.NewReader(out)) + for scanner.Scan() { + line := scanner.Text() + trimmed := strings.TrimSpace(line) + + switch { + case strings.HasPrefix(trimmed, "pool:"): + pools = append(pools, PoolStatus{Name: strings.TrimSpace(strings.TrimPrefix(trimmed, "pool:"))}) + current = &pools[len(pools)-1] + inConfig = false + scanContinuation = false + case current == nil: + continue + case strings.HasPrefix(trimmed, "state:"): + current.State = strings.TrimSpace(strings.TrimPrefix(trimmed, "state:")) + case strings.HasPrefix(trimmed, "scan:"): + current.Scrub = parseScanLine(trimmed) + // zpool status prints the progress percentage on the line after scan. + scanContinuation = true + case trimmed == "config:": + inConfig = true + case scanContinuation: + // The line after scan: may be an indented progress continuation. + if m := progressRe.FindStringSubmatch(trimmed); m != nil { + current.Scrub.Progress = m[1] + "%" + } + scanContinuation = false + case inConfig && (line == "" || strings.HasPrefix(line, " ") || strings.HasPrefix(line, "\t")): + // Table rows are indented; blank lines separate sections. The + // column header and the pool's own row are skipped. + if trimmed != "" && !strings.HasPrefix(trimmed, "NAME") { + if vdev, ok := parseVdevLine(trimmed, current.Name); ok { + current.Vdevs = append(current.Vdevs, vdev) + } + } + case inConfig: + // unindented line (errors:, status:, next pool:) ends the table + inConfig = false + } + } + return pools, scanner.Err() +} + +// parseScanLine maps a `scan:` line to a ScrubStatus. +func parseScanLine(line string) ScrubStatus { + var scrub ScrubStatus + switch { + case strings.Contains(line, "in progress"): + scrub.State = "SCANNING" + case strings.Contains(line, "canceled"): + scrub.State = "CANCELED" + case strings.Contains(line, "repaired"), strings.Contains(line, "resilvered"): + scrub.State = "FINISHED" + default: + scrub.State = "NONE" + } + if m := progressRe.FindStringSubmatch(line); m != nil { + scrub.Progress = m[1] + "%" + } + if m := errorsRe.FindStringSubmatch(line); m != nil { + if n, err := strconv.ParseUint(m[1], 10, 64); err == nil { + scrub.Errors = n + } + } + return scrub +} + +// parseVdevLine parses one row of the config table. Rows have the shape +// "NAME STATE READ WRITE CKSUM [extra...]". The first data row is the pool +// itself and is skipped since it duplicates pool-level info. +func parseVdevLine(line, poolName string) (VdevStatus, bool) { + fields := strings.Fields(line) + if len(fields) < 5 { + return VdevStatus{}, false + } + if fields[0] == poolName { + return VdevStatus{}, false + } + read, err1 := strconv.ParseUint(fields[2], 10, 64) + write, err2 := strconv.ParseUint(fields[3], 10, 64) + cksum, err3 := strconv.ParseUint(fields[4], 10, 64) + if err1 != nil || err2 != nil || err3 != nil { + return VdevStatus{}, false + } + return VdevStatus{ + Name: fields[0], + State: fields[1], + ReadErrs: read, + WriteErrs: write, + ChecksumErrs: cksum, + }, true +} diff --git a/agent/zfs/zfs_test.go b/agent/zfs/zfs_test.go new file mode 100644 index 00000000..0472434d --- /dev/null +++ b/agent/zfs/zfs_test.go @@ -0,0 +1,143 @@ +//go:build testing + +package zfs + +import ( + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func fixturePath(name string) string { + return filepath.Join("..", "test-data", "zfs", name) +} + +func TestParseZpoolListOutput(t *testing.T) { + data, err := os.ReadFile(fixturePath("zpool_list.txt")) + require.NoError(t, err) + + pools, err := parseZpoolListOutput(data) + require.NoError(t, err) + require.Len(t, pools, 2) + + assert.Equal(t, PoolStat{Name: "tank", Size: 23999000000000, Alloc: 12000000000000, Free: 11999000000000, Health: "ONLINE"}, pools[0]) + assert.Equal(t, PoolStat{Name: "rpool", Size: 1200000000000, Alloc: 900000000000, Free: 300000000000, Health: "DEGRADED"}, pools[1]) +} + +func TestParseZpoolListOutputIgnoresEmptyLines(t *testing.T) { + pools, err := parseZpoolListOutput([]byte("tank\t100\t50\t50\tONLINE\n\n")) + require.NoError(t, err) + require.Len(t, pools, 1) + assert.Equal(t, "tank", pools[0].Name) +} + +func TestParseZpoolListOutputNoPools(t *testing.T) { + pools, err := parseZpoolListOutput([]byte("no pools available\n")) + require.NoError(t, err) + assert.Empty(t, pools) +} + +func TestParseZpoolListOutputRejectsMalformedLine(t *testing.T) { + _, err := parseZpoolListOutput([]byte("tank\t100\t50\n")) + require.Error(t, err) + + _, err = parseZpoolListOutput([]byte("tank\tnotanumber\t50\t50\tONLINE\n")) + require.Error(t, err) +} + +func TestParseZfsListOutput(t *testing.T) { + data, err := os.ReadFile(fixturePath("zfs_list.txt")) + require.NoError(t, err) + + datasets, err := parseZfsListOutput(data) + require.NoError(t, err) + require.Len(t, datasets, 9) + + // Mountpoint with a space must be kept intact (tab-split only). + assert.Equal(t, "/tank/my media", datasets[3].Mountpoint) + // Unmounted datasets/zvols report "-". + assert.Equal(t, "-", datasets[4].Mountpoint) + assert.Equal(t, uint64(12000000000000), datasets[0].Used) + assert.Equal(t, uint64(11999000000000), datasets[0].Avail) +} + +func TestParseZpoolStatusOutput(t *testing.T) { + data, err := os.ReadFile(fixturePath("zpool_status.txt")) + require.NoError(t, err) + + pools, err := parseZpoolStatusOutput(data) + require.NoError(t, err) + require.Len(t, pools, 2) + + tank := pools[0] + assert.Equal(t, "tank", tank.Name) + assert.Equal(t, "ONLINE", tank.State) + assert.Equal(t, "FINISHED", tank.Scrub.State) + assert.Equal(t, "", tank.Scrub.Progress) + assert.Equal(t, uint64(0), tank.Scrub.Errors) + // Pool row itself is skipped; mirror + 2 disks remain. + require.Len(t, tank.Vdevs, 3) + assert.Equal(t, "mirror-0", tank.Vdevs[0].Name) + assert.Equal(t, "sda", tank.Vdevs[1].Name) + assert.Equal(t, "sdb", tank.Vdevs[2].Name) + + rpool := pools[1] + assert.Equal(t, "rpool", rpool.Name) + assert.Equal(t, "DEGRADED", rpool.State) + assert.Equal(t, "SCANNING", rpool.Scrub.State) + assert.Equal(t, "10.00%", rpool.Scrub.Progress) + require.Len(t, rpool.Vdevs, 3) + assert.Equal(t, "FAULTED", rpool.Vdevs[2].State) + assert.Equal(t, uint64(1), rpool.Vdevs[2].ReadErrs) + assert.Equal(t, uint64(2), rpool.Vdevs[2].WriteErrs) + assert.Equal(t, uint64(3), rpool.Vdevs[2].ChecksumErrs) +} + +func TestParseScanLine(t *testing.T) { + assert.Equal(t, "FINISHED", parseScanLine("scan: scrub repaired 0B in 00:05:12 with 0 errors on Sun Jun 1 02:00:12 2025").State) + assert.Equal(t, uint64(3), parseScanLine("scan: scrub repaired 10G in 01:00:00 with 3 errors on Sun Jun 1 02:00:12 2025").Errors) + assert.Equal(t, "SCANNING", parseScanLine("scan: scrub in progress since Sun Jun 8 01:00:00 2025").State) + assert.Equal(t, "CANCELED", parseScanLine("scan: scrub canceled on Sun Jun 1 02:00:12 2025").State) + assert.Equal(t, "FINISHED", parseScanLine("scan: resilvered 1.23G in 00:01:00 with 0 errors on Sun Jun 1 02:00:12 2025").State) + assert.Equal(t, "NONE", parseScanLine("scan: none requested").State) +} + +func TestCommandOutputForcesLocaleAndTimesOut(t *testing.T) { + t.Setenv("BESZEL_ZFS_COMMAND_HELPER", "1") + out, err := commandOutput(os.Args[0], "-test.run=TestZfsCommandHelperProcess", "--", "locale") + require.NoError(t, err) + assert.Equal(t, "C/C", string(out)) + + oldTimeout := commandTimeout + commandTimeout = 20 * time.Millisecond + t.Cleanup(func() { commandTimeout = oldTimeout }) + _, err = commandOutput(os.Args[0], "-test.run=TestZfsCommandHelperProcess", "--", "sleep") + require.Error(t, err) + assert.Contains(t, err.Error(), "timed out") +} + +func TestZfsCommandHelperProcess(t *testing.T) { + if os.Getenv("BESZEL_ZFS_COMMAND_HELPER") != "1" { + return + } + mode := "" + for i, arg := range os.Args { + if arg == "--" && i+1 < len(os.Args) { + mode = os.Args[i+1] + break + } + } + switch strings.TrimSpace(mode) { + case "locale": + _, _ = fmt.Printf("%s/%s", os.Getenv("LC_ALL"), os.Getenv("LANG")) + case "sleep": + time.Sleep(time.Second) + } + os.Exit(0) +} diff --git a/agent/zfs/zfs_unsupported.go b/agent/zfs/zfs_unsupported.go index 95aa8deb..0e9d3abc 100644 --- a/agent/zfs/zfs_unsupported.go +++ b/agent/zfs/zfs_unsupported.go @@ -7,3 +7,7 @@ import "errors" func ARCSize() (uint64, error) { return 0, errors.ErrUnsupported } + +func PoolKernelStats() ([]PoolKernelStat, error) { + return nil, errors.ErrUnsupported +} diff --git a/agent/zfs_pool.go b/agent/zfs_pool.go new file mode 100644 index 00000000..12988711 --- /dev/null +++ b/agent/zfs_pool.go @@ -0,0 +1,320 @@ +package agent + +import ( + "log/slog" + "strings" + "sync" + "time" + + "github.com/henrygd/beszel/agent/zfs" + "github.com/henrygd/beszel/internal/entities/system" + zfsentity "github.com/henrygd/beszel/internal/entities/zfs" +) + +// zfsDatasetUsage holds usage values for a ZFS dataset mountpoint. +type zfsDatasetUsage struct { + used uint64 + avail uint64 +} + +// datasetUsageRefreshInterval controls how often `zfs list` is re-run for the +// mountpoint usage map. Dataset inventory changes rarely. +const datasetUsageRefreshInterval = 5 * time.Minute + +// poolStatsRefreshInterval controls how often `zpool list` is re-run for pool +// capacity. Health and I/O are read from procfs on Linux, so the utility only +// needs to refresh slow-moving space accounting. +const poolStatsRefreshInterval = time.Minute + +type poolKernelSample struct { + nread uint64 + nwrite uint64 + at time.Time +} + +// ZfsManager collects ZFS pool and dataset statistics. Collection functions +// are fields so unit tests can substitute them (same pattern as +// diskDiscovery.usageFn). It is safe for concurrent use by a single goroutine +// only; callers must hold the agent lock like updateDiskUsage does. +type ZfsManager struct { + poolStatsFn func() ([]zfs.PoolStat, error) // capacity/health source + datasetsFn func() ([]zfs.Dataset, error) // dataset inventory source + kernelStatsFn func() ([]zfs.PoolKernelStat, error) // procfs pool state/I/O source + poolStatusesFn func() ([]zfs.PoolStatus, error) // scrub/vdev detail source + + poolData []zfs.PoolStat // cached pool inventory (TTL below) + lastPoolStats time.Time + kernelSamples map[string]poolKernelSample + + datasetUsage map[string]zfsDatasetUsage // mountpoint -> usage + lastUsageRefresh time.Time + + // Detail data (pools, vdevs, scrub, datasets) is cached and refreshed on + // an interval. Accessed from handler goroutines, so it is mutex-protected. + detailMu sync.Mutex + detail *zfsentity.ZfsData + lastDetailRefresh time.Time + detailInterval time.Duration +} + +// newZfsManager creates a ZfsManager wired to the system's ZFS utilities. +func newZfsManager() *ZfsManager { + return &ZfsManager{ + poolStatsFn: zfs.PoolStats, + datasetsFn: zfs.Datasets, + kernelStatsFn: zfs.PoolKernelStats, + poolStatusesFn: zfs.PoolStatuses, + detailInterval: time.Hour, + } +} + +// Update refreshes systemStats.ZfsPools with the latest pool data. I/O +// throughput and health come from inexpensive kernel kstats on Linux. Pool +// capacity and dataset usage come from separately cached utility calls. It is +// a no-op when ZFS is absent. +func (zm *ZfsManager) Update(systemStats *system.Stats) { + pools := zm.poolStats() + if len(pools) == 0 { + return + } + + kernelStats, ioRates := zm.kernelStats() + + if systemStats.ZfsPools == nil { + systemStats.ZfsPools = make(map[string]*system.ZfsPool, len(pools)) + } + for i := range pools { + pool := &pools[i] + // Full precision, matching the dataset values below; the frontend + // formats any magnitude. + stats := &system.ZfsPool{ + Total: float64(pool.Size) / (1024 * 1024 * 1024), + Used: float64(pool.Alloc) / (1024 * 1024 * 1024), + Health: pool.Health, + } + if kernel, exists := kernelStats[pool.Name]; exists && kernel.Health != "" { + stats.Health = kernel.Health + } + if io, exists := ioRates[pool.Name]; exists { + stats.ReadBytes = io.NRead + stats.WriteBytes = io.NWrite + } + slog.Debug("ZFS pool sample", "pool", pool.Name, "health", stats.Health, "used_gb", stats.Used, "read_bps", stats.ReadBytes, "write_bps", stats.WriteBytes) + systemStats.ZfsPools[pool.Name] = stats + } + +} + +// poolStats returns the cached pool inventory, re-running `zpool list` at most +// every poolStatsRefreshInterval. On failure the previous inventory is +// retained and the refresh is retried on the next cadence. +func (zm *ZfsManager) poolStats() []zfs.PoolStat { + if zm.lastPoolStats.IsZero() || time.Since(zm.lastPoolStats) >= poolStatsRefreshInterval { + pools, err := zm.poolStatsFn() + if err != nil { + slog.Debug("ZFS pool stats unavailable", "err", err) + } else { + zm.poolData = pools + } + zm.lastPoolStats = time.Now() + } + return zm.poolData +} + +// kernelStats reads cumulative pool counters and converts them to per-second +// rates. Counter decreases indicate a pool export/import and reset the +// baseline instead of producing an underflow spike. +func (zm *ZfsManager) kernelStats() (map[string]zfs.PoolKernelStat, map[string]zfs.PoolIoStats) { + if zm.kernelStatsFn == nil { + return nil, nil + } + stats, err := zm.kernelStatsFn() + if err != nil { + slog.Debug("ZFS kernel stats unavailable", "err", err) + return nil, nil + } + now := time.Now() + byName := make(map[string]zfs.PoolKernelStat, len(stats)) + rates := make(map[string]zfs.PoolIoStats, len(stats)) + nextSamples := make(map[string]poolKernelSample, len(stats)) + for _, stat := range stats { + byName[stat.Name] = stat + if previous, ok := zm.kernelSamples[stat.Name]; ok && now.After(previous.at) && + stat.NRead >= previous.nread && stat.NWrite >= previous.nwrite { + seconds := now.Sub(previous.at).Seconds() + rates[stat.Name] = zfs.PoolIoStats{ + NRead: uint64(float64(stat.NRead-previous.nread) / seconds), + NWrite: uint64(float64(stat.NWrite-previous.nwrite) / seconds), + } + } + nextSamples[stat.Name] = poolKernelSample{nread: stat.NRead, nwrite: stat.NWrite, at: now} + } + zm.kernelSamples = nextSamples + return byName, rates +} + +// refreshDatasetUsage re-runs `zfs list` when the refresh window has elapsed +// and rebuilds the mountpoint-keyed usage map. +func (zm *ZfsManager) refreshDatasetUsage() { + if !zm.lastUsageRefresh.IsZero() && time.Since(zm.lastUsageRefresh) < datasetUsageRefreshInterval { + return + } + datasets, err := zm.datasetsFn() + if err != nil { + slog.Debug("ZFS dataset usage unavailable", "err", err) + } else { + usage := make(map[string]zfsDatasetUsage, len(datasets)) + for _, ds := range datasets { + if ds.Mountpoint != "" && ds.Mountpoint != "-" { + usage[ds.Mountpoint] = zfsDatasetUsage{used: ds.Used, avail: ds.Avail} + } + } + zm.datasetUsage = usage + } + zm.lastUsageRefresh = time.Now() +} + +// DatasetUsage returns ZFS dataset usage keyed by mountpoint, refreshed at +// most every datasetUsageRefreshInterval. On failure the previous map is +// retained and a debug log is emitted. +func (zm *ZfsManager) DatasetUsage() map[string]zfsDatasetUsage { + zm.refreshDatasetUsage() + return zm.datasetUsage +} + +// GetDetail returns ZFS detail data (pool health, scrub, vdevs, datasets). +// Scheduled requests use the cached snapshot until stale; manual requests can +// force collection. On failure the previous snapshot is retained. +func (zm *ZfsManager) GetDetail(force bool) *zfsentity.ZfsData { + zm.detailMu.Lock() + defer zm.detailMu.Unlock() + + if force || zm.detail == nil || time.Since(zm.lastDetailRefresh) >= zm.detailInterval { + if data, err := zm.collectDetail(zm.detail); err != nil { + slog.Debug("ZFS detail collection failed", "err", err) + if zm.detail == nil { + return &zfsentity.ZfsData{} + } + return &zfsentity.ZfsData{Pools: zm.detail.Pools} + } else { + zm.detail = data + zm.lastDetailRefresh = time.Now() + } + } + if zm.detail == nil { + return &zfsentity.ZfsData{} + } + return zm.detail +} + +// collectDetail builds a ZfsData payload from the current system state. +func (zm *ZfsManager) collectDetail(previous *zfsentity.ZfsData) (*zfsentity.ZfsData, error) { + pools, err := zm.poolStatsFn() + if err != nil { + return nil, err + } + if len(pools) == 0 { + return &zfsentity.ZfsData{Pools: []*zfsentity.PoolDetail{}, Complete: true}, nil + } + + statuses, statusErr := zm.poolStatusesFn() + if statusErr != nil { + slog.Debug("ZFS pool status unavailable", "err", statusErr) + } + datasets, datasetsErr := zm.datasetsFn() + if datasetsErr != nil { + slog.Debug("ZFS datasets unavailable", "err", datasetsErr) + } + + statusByPool := make(map[string]zfs.PoolStatus, len(statuses)) + for _, st := range statuses { + statusByPool[st.Name] = st + } + + previousByPool := make(map[string]*zfsentity.PoolDetail) + if previous != nil { + for _, pool := range previous.Pools { + if pool != nil { + previousByPool[pool.Name] = pool + } + } + } + + data := &zfsentity.ZfsData{Pools: make([]*zfsentity.PoolDetail, 0, len(pools)), Complete: true} + for i := range pools { + p := &pools[i] + detail := &zfsentity.PoolDetail{ + Name: p.Name, + Health: p.Health, + Size: p.Size, + Alloc: p.Alloc, + Free: p.Free, + } + if st, ok := statusByPool[p.Name]; statusErr == nil && ok { + if st.Scrub.State != "" && st.Scrub.State != "NONE" { + detail.Scrub = &zfsentity.Scrub{ + State: st.Scrub.State, + Progress: st.Scrub.Progress, + Errors: st.Scrub.Errors, + } + } + for _, v := range st.Vdevs { + detail.Vdevs = append(detail.Vdevs, &zfsentity.Vdev{ + Name: v.Name, + State: v.State, + ReadErrs: v.ReadErrs, + WriteErrs: v.WriteErrs, + ChecksumErrs: v.ChecksumErrs, + }) + } + } else { + if cached := previousByPool[p.Name]; cached != nil { + detail.Scrub = cached.Scrub + detail.Vdevs = cached.Vdevs + } + } + if datasetsErr == nil { + foundDataset := false + for _, ds := range datasets { + if poolOfDataset(ds.Name) == p.Name { + foundDataset = true + detail.Datasets = append(detail.Datasets, &zfsentity.Dataset{ + Name: ds.Name, + Used: ds.Used, + Avail: ds.Avail, + Mountpoint: ds.Mountpoint, + }) + } + } + if !foundDataset { + if cached := previousByPool[p.Name]; cached != nil { + detail.Datasets = cached.Datasets + } + } + } else if cached := previousByPool[p.Name]; cached != nil { + detail.Datasets = cached.Datasets + } + data.Pools = append(data.Pools, detail) + } + return data, nil +} + +// poolOfDataset returns the pool name for a dataset name (everything before +// the first '/'). Datasets without a separator belong to a pool of the same +// name. +func poolOfDataset(name string) string { + if idx := strings.IndexByte(name, '/'); idx >= 0 { + return name[:idx] + } + return name +} + +// ZfsMountpoints returns the set of mountpoints backed by ZFS datasets. +func (zm *ZfsManager) ZfsMountpoints() map[string]bool { + usage := zm.DatasetUsage() + mountpoints := make(map[string]bool, len(usage)) + for mountpoint := range usage { + mountpoints[mountpoint] = true + } + return mountpoints +} diff --git a/agent/zfs_pool_test.go b/agent/zfs_pool_test.go new file mode 100644 index 00000000..7101ae89 --- /dev/null +++ b/agent/zfs_pool_test.go @@ -0,0 +1,245 @@ +//go:build testing + +package agent + +import ( + "testing" + "time" + + "github.com/henrygd/beszel/agent/zfs" + "github.com/henrygd/beszel/internal/entities/system" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestUpdatePopulatesZfsPools(t *testing.T) { + zm := &ZfsManager{} + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { + return []zfs.PoolStat{{Name: "tank", Size: 23999000000000, Alloc: 12000000000000, Free: 11999000000000, Health: "DEGRADED"}}, nil + } + zm.datasetsFn = func() ([]zfs.Dataset, error) { + return []zfs.Dataset{ + {Name: "tank/apps", Used: 5000000000000, Avail: 11999000000000, Mountpoint: "/tank/apps"}, + {Name: "tank/backup", Used: 6000000000000, Avail: 11999000000000, Mountpoint: "/tank/backup"}, + // Small zvol (Proxmox VM EFI disk): must not round to zero. + {Name: "rpool/vm-100-disk-2", Used: 4194304, Avail: 0, Mountpoint: "-"}, + }, nil + } + var kernelCalls int + zm.kernelStatsFn = func() ([]zfs.PoolKernelStat, error) { + kernelCalls++ + return []zfs.PoolKernelStat{{ + Name: "tank", Health: "ONLINE", + NRead: uint64(kernelCalls-1) * 1250, NWrite: uint64(kernelCalls-1) * 5120, + }}, nil + } + + var stats system.Stats + // The first kernel sample establishes the cumulative-counter baseline. + zm.Update(&stats) + zm.kernelSamples["tank"] = poolKernelSample{at: time.Now().Add(-time.Second)} + zm.Update(&stats) + require.NotNil(t, stats.ZfsPools) + require.Contains(t, stats.ZfsPools, "tank") + assert.InDelta(t, 22350.8105, stats.ZfsPools["tank"].Total, 0.0001) // Size in GiB + assert.InDelta(t, 11175.8709, stats.ZfsPools["tank"].Used, 0.0001) // Alloc in GiB + assert.Equal(t, "ONLINE", stats.ZfsPools["tank"].Health) + assert.InDelta(t, 1250, stats.ZfsPools["tank"].ReadBytes, 5) + assert.InDelta(t, 5120, stats.ZfsPools["tank"].WriteBytes, 5) + +} + +// TestUpdateKernelStatsMissing verifies pools without a kernel sample report zero +// I/O instead of erroring. +func TestUpdateKernelStatsMissing(t *testing.T) { + zm := &ZfsManager{} + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { + return []zfs.PoolStat{{Name: "tank", Size: 1, Alloc: 1, Health: "ONLINE"}}, nil + } + zm.datasetsFn = func() ([]zfs.Dataset, error) { return nil, nil } + zm.kernelStatsFn = func() ([]zfs.PoolKernelStat, error) { + return nil, zfs.ErrNoZfs + } + + var stats system.Stats + zm.Update(&stats) + require.NotNil(t, stats.ZfsPools) + assert.Equal(t, uint64(0), stats.ZfsPools["tank"].ReadBytes) + assert.Equal(t, uint64(0), stats.ZfsPools["tank"].WriteBytes) +} + +func TestUpdateKernelCounterReset(t *testing.T) { + zm := &ZfsManager{} + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { + return []zfs.PoolStat{{Name: "tank", Health: "ONLINE"}}, nil + } + zm.datasetsFn = func() ([]zfs.Dataset, error) { return nil, nil } + zm.kernelSamples = map[string]poolKernelSample{ + "tank": {nread: 100, nwrite: 200, at: time.Now().Add(-time.Second)}, + } + zm.kernelStatsFn = func() ([]zfs.PoolKernelStat, error) { + return []zfs.PoolKernelStat{{Name: "tank", Health: "ONLINE", NRead: 10, NWrite: 20}}, nil + } + + var stats system.Stats + zm.Update(&stats) + assert.Equal(t, uint64(0), stats.ZfsPools["tank"].ReadBytes) + assert.Equal(t, uint64(0), stats.ZfsPools["tank"].WriteBytes) +} + +func TestUpdateNoZfs(t *testing.T) { + zm := &ZfsManager{} + calls := 0 + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { + calls++ + return nil, zfs.ErrNoZfs + } + + var stats system.Stats + zm.Update(&stats) + zm.Update(&stats) + assert.Nil(t, stats.ZfsPools) + assert.Equal(t, 1, calls, "failed pool discovery should be cached until the next refresh interval") +} + +func TestUpdateEmptyPools(t *testing.T) { + zm := &ZfsManager{} + calls := 0 + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { + calls++ + return nil, nil + } + + var stats system.Stats + zm.Update(&stats) + zm.Update(&stats) + assert.Nil(t, stats.ZfsPools) + assert.Equal(t, 1, calls, "an empty pool inventory should be cached until the next refresh interval") +} + +func TestDatasetUsage(t *testing.T) { + zm := &ZfsManager{} + calls := 0 + zm.datasetsFn = func() ([]zfs.Dataset, error) { + calls++ + return []zfs.Dataset{ + {Name: "tank", Used: 12000000000000, Avail: 11999000000000, Mountpoint: "/tank"}, + {Name: "tank/apps", Used: 1000000000000, Avail: 11999000000000, Mountpoint: "/tank/apps"}, + {Name: "rpool", Used: 900000000000, Avail: 300000000000, Mountpoint: "-"}, // zvol/unmounted: excluded + }, nil + } + + usage := zm.DatasetUsage() + require.Len(t, usage, 2) + assert.Equal(t, zfsDatasetUsage{used: 12000000000000, avail: 11999000000000}, usage["/tank"]) + assert.Equal(t, zfsDatasetUsage{used: 1000000000000, avail: 11999000000000}, usage["/tank/apps"]) + assert.Equal(t, 1, calls) + + // Second call within the refresh window must not re-run the collector. + zm.DatasetUsage() + assert.Equal(t, 1, calls) +} + +func TestDatasetUsageRefreshOnErrorKeepsPrevious(t *testing.T) { + zm := &ZfsManager{} + zm.datasetsFn = func() ([]zfs.Dataset, error) { + return []zfs.Dataset{{Name: "tank", Used: 1, Avail: 1, Mountpoint: "/tank"}}, nil + } + assert.Len(t, zm.DatasetUsage(), 1) + + // Force refresh window expiry, then a failing collector. + zm.lastUsageRefresh = time.Now().Add(-10 * time.Minute) + zm.datasetsFn = func() ([]zfs.Dataset, error) { + return nil, zfs.ErrNoZfs + } + usage := zm.DatasetUsage() + assert.Len(t, usage, 1, "previous usage should be retained on error") +} + +func TestGetDetailForceRefresh(t *testing.T) { + zm := &ZfsManager{detailInterval: time.Hour} + poolCalls := 0 + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { + poolCalls++ + return []zfs.PoolStat{{Name: "tank", Alloc: uint64(poolCalls)}}, nil + } + zm.poolStatusesFn = func() ([]zfs.PoolStatus, error) { return nil, nil } + zm.datasetsFn = func() ([]zfs.Dataset, error) { return nil, nil } + + first := zm.GetDetail(false) + assert.True(t, first.Complete) + require.Len(t, first.Pools, 1) + assert.Equal(t, uint64(1), first.Pools[0].Alloc) + + cached := zm.GetDetail(false) + require.Len(t, cached.Pools, 1) + assert.Equal(t, uint64(1), cached.Pools[0].Alloc) + assert.Equal(t, 1, poolCalls) + + refreshed := zm.GetDetail(true) + assert.True(t, refreshed.Complete) + require.Len(t, refreshed.Pools, 1) + assert.Equal(t, uint64(2), refreshed.Pools[0].Alloc) + assert.Equal(t, 2, poolCalls) +} + +func TestGetDetailSuccessfulEmptyInventoryClearsCache(t *testing.T) { + zm := &ZfsManager{detailInterval: time.Hour} + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { + return []zfs.PoolStat{{Name: "tank"}}, nil + } + zm.poolStatusesFn = func() ([]zfs.PoolStatus, error) { return nil, nil } + zm.datasetsFn = func() ([]zfs.Dataset, error) { return nil, nil } + + require.Len(t, zm.GetDetail(false).Pools, 1) + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { return nil, nil } + empty := zm.GetDetail(true) + assert.True(t, empty.Complete) + assert.Empty(t, empty.Pools) +} + +func TestGetDetailFailureReturnsIncompleteCachedInventory(t *testing.T) { + zm := &ZfsManager{detailInterval: time.Hour} + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { + return []zfs.PoolStat{{Name: "tank"}}, nil + } + zm.poolStatusesFn = func() ([]zfs.PoolStatus, error) { + return []zfs.PoolStatus{{Name: "tank", Vdevs: []zfs.VdevStatus{{Name: "mirror-0"}}}}, nil + } + zm.datasetsFn = func() ([]zfs.Dataset, error) { + return []zfs.Dataset{{Name: "tank/data"}}, nil + } + first := zm.GetDetail(false) + require.True(t, first.Complete) + require.Len(t, first.Pools[0].Vdevs, 1) + require.Len(t, first.Pools[0].Datasets, 1) + + zm.poolStatusesFn = func() ([]zfs.PoolStatus, error) { return nil, zfs.ErrNoZfs } + zm.datasetsFn = func() ([]zfs.Dataset, error) { return nil, zfs.ErrNoZfs } + partial := zm.GetDetail(true) + require.True(t, partial.Complete) + require.Len(t, partial.Pools[0].Vdevs, 1) + require.Len(t, partial.Pools[0].Datasets, 1) + + zm.poolStatsFn = func() ([]zfs.PoolStat, error) { return nil, zfs.ErrNoZfs } + lastSuccessfulRefresh := zm.lastDetailRefresh + failed := zm.GetDetail(true) + assert.False(t, failed.Complete) + require.Len(t, failed.Pools, 1) + assert.Equal(t, "tank", failed.Pools[0].Name) + assert.Equal(t, lastSuccessfulRefresh, zm.lastDetailRefresh) +} + +func TestZfsMountpoints(t *testing.T) { + zm := &ZfsManager{} + zm.datasetsFn = func() ([]zfs.Dataset, error) { + return []zfs.Dataset{ + {Name: "tank", Mountpoint: "/tank"}, + {Name: "rpool/ROOT/pve-1", Mountpoint: "/"}, + }, nil + } + mountpoints := zm.ZfsMountpoints() + assert.Len(t, mountpoints, 2) + assert.True(t, mountpoints["/tank"]) + assert.True(t, mountpoints["/"]) +} diff --git a/beszel.go b/beszel.go index a288c0f6..82e87e81 100644 --- a/beszel.go +++ b/beszel.go @@ -6,7 +6,7 @@ import "github.com/blang/semver" const ( // Version is the current version of the application. - Version = "0.18.8" + Version = "0.18.9" // AppName is the name of the application. AppName = "beszel" ) @@ -16,3 +16,6 @@ var MinVersionCbor = semver.MustParse("0.12.0") // MinVersionAgentResponse is the minimum supported version for AgentResponse compatibility. var MinVersionAgentResponse = semver.MustParse("0.13.0") + +// MinVersionZfsData is the minimum agent version that supports ZFS detail requests. +var MinVersionZfsData = semver.MustParse("0.18.9") diff --git a/internal/alerts/alerts.go b/internal/alerts/alerts.go index 826f4d68..6dc57626 100644 --- a/internal/alerts/alerts.go +++ b/internal/alerts/alerts.go @@ -57,12 +57,18 @@ type SystemAlertStats struct { Battery [2]uint8 `json:"bat"` Batteries map[string]uint8 `json:"bats"` ExtraFs map[string]SystemAlertFsStats `json:"efs"` + ZfsPools map[string]SystemAlertZfsPool `json:"z"` } type SystemAlertGPUData struct { Usage float64 `json:"u"` } +type SystemAlertZfsPool struct { + Total float64 `json:"d"` + Used float64 `json:"du"` +} + type SystemAlertData struct { systemRecord *core.Record alertData CachedAlertData @@ -111,6 +117,9 @@ func (am *AlertManager) bindEvents() { am.hub.OnRecordAfterUpdateSuccess("alerts").BindFunc(updateHistoryOnAlertUpdate) am.hub.OnRecordAfterDeleteSuccess("alerts").BindFunc(resolveHistoryOnAlertDelete) am.hub.OnRecordAfterUpdateSuccess("smart_devices").BindFunc(am.handleSmartDeviceAlert) + am.hub.OnRecordAfterCreateSuccess("zfs_pools").BindFunc(am.handleZfsPoolCreateAlert) + am.hub.OnRecordAfterUpdateSuccess("zfs_pools").BindFunc(am.handleZfsPoolAlert) + am.hub.OnRecordAfterDeleteSuccess("zfs_pools").BindFunc(resolveZfsPoolHistoryOnDelete) am.hub.OnServe().BindFunc(func(e *core.ServeEvent) error { // Populate all alerts into cache on startup diff --git a/internal/alerts/alerts_system.go b/internal/alerts/alerts_system.go index 04001d98..690a6397 100644 --- a/internal/alerts/alerts_system.go +++ b/internal/alerts/alerts_system.go @@ -44,6 +44,14 @@ func (am *AlertManager) HandleSystemAlerts(systemRecord *core.Record, data *syst maxUsedPct = usedPct } } + for _, pool := range data.Stats.ZfsPools { + if pool != nil && pool.Total > 0 { + usedPct := pool.Used / pool.Total * 100 + if usedPct > maxUsedPct { + maxUsedPct = usedPct + } + } + } val = maxUsedPct case "Temperature": if data.Info.DashboardTemp < 1 { @@ -208,6 +216,16 @@ func (am *AlertManager) HandleSystemAlerts(systemRecord *core.Record, data *syst alert.mapSums[key] += float32(fs.DiskUsed / fs.DiskTotal * 100) } } + // add zfs pool usage from historical record + for key, pool := range stats.ZfsPools { + if pool.Total > 0 { + zfsKey := zfsDiskAlertKey(key) + if _, ok := alert.mapSums[zfsKey]; !ok { + alert.mapSums[zfsKey] = 0.0 + } + alert.mapSums[zfsKey] += float32(pool.Used / pool.Total * 100) + } + } case "Temperature": if alert.mapSums == nil { alert.mapSums = make(map[string]float32, len(stats.Temperatures)) @@ -255,7 +273,7 @@ func (am *AlertManager) HandleSystemAlerts(systemRecord *core.Record, data *syst sumPct := float32(value) if sumPct > maxPct { maxPct = sumPct - alert.descriptor = fmt.Sprintf("Usage of %s", key) + alert.descriptor = diskAlertDescriptor(key) } } alert.val = float64(maxPct / float32(alert.count)) @@ -301,6 +319,17 @@ func (am *AlertManager) HandleSystemAlerts(systemRecord *core.Record, data *syst return nil } +func zfsDiskAlertKey(poolName string) string { + return "zfs:" + poolName +} + +func diskAlertDescriptor(key string) string { + if poolName, ok := strings.CutPrefix(key, "zfs:"); ok { + return fmt.Sprintf("Usage of ZFS pool %s", poolName) + } + return fmt.Sprintf("Usage of %s", key) +} + func hasRepresentativeBattery(legacy [2]uint8, batteries map[string]uint8) bool { return legacy != [2]uint8{} || len(batteries) > 0 } diff --git a/internal/alerts/alerts_zfs.go b/internal/alerts/alerts_zfs.go new file mode 100644 index 00000000..f3591f46 --- /dev/null +++ b/internal/alerts/alerts_zfs.go @@ -0,0 +1,142 @@ +package alerts + +import ( + "fmt" + "time" + + "github.com/pocketbase/dbx" + "github.com/pocketbase/pocketbase/core" +) + +// handleZfsPoolAlert sends alerts when a ZFS pool health state worsens and +// resolves the alert history entry when the pool recovers. Like the SMART +// hook, this is automatic and does not require user opt-in. +func (am *AlertManager) handleZfsPoolAlert(e *core.RecordEvent) error { + return am.handleZfsPoolHealthAlert(e, e.Record.Original().GetString("health")) +} + +func (am *AlertManager) handleZfsPoolCreateAlert(e *core.RecordEvent) error { + return am.handleZfsPoolHealthAlert(e, "") +} + +func (am *AlertManager) handleZfsPoolHealthAlert(e *core.RecordEvent, oldHealth string) error { + newHealth := e.Record.GetString("health") + oldSeverity := zfsPoolSeverity(oldHealth) + newSeverity := zfsPoolSeverity(newHealth) + + systemID := e.Record.GetString("system") + if systemID == "" { + return e.Next() + } + + systemRecord, err := e.App.FindRecordById("systems", systemID) + if err != nil { + e.App.Logger().Error("Failed to find system for ZFS alert", "err", err, "systemID", systemID) + return e.Next() + } + + // Pool recovered to a healthy state: resolve any open history entries. + if newSeverity == 1 && oldSeverity > 1 { + resolveAllAlertHistoryRecords(e.App, e.Record.Id) + return e.Next() + } + + if !shouldSendZfsPoolAlert(oldSeverity, newSeverity) { + return e.Next() + } + + systemName := systemRecord.GetString("name") + poolName := e.Record.GetString("name") + + title := fmt.Sprintf("ZFS pool %s on %s: %s", newHealth, systemName, poolName) + message := fmt.Sprintf("ZFS pool %s (%s) was first observed as %s", poolName, systemName, newHealth) + if oldSeverity > 0 { + message = fmt.Sprintf("ZFS pool %s (%s) health changed from %s to %s", poolName, systemName, oldHealth, newHealth) + } + + userIDs := systemRecord.GetStringSlice("users") + if len(userIDs) == 0 { + return e.Next() + } + + for _, userID := range userIDs { + if err := am.SendAlert(AlertMessageData{ + UserID: userID, + SystemID: systemID, + Title: title, + Message: message, + Link: am.hub.MakeLink("system", systemID), + LinkText: "View " + systemName, + }); err != nil { + e.App.Logger().Error("Failed to send ZFS alert", "err", err, "userID", userID) + } + _ = createZfsPoolHistoryRecord(e.App, userID, systemID, e.Record.Id, poolName) + } + + return e.Next() +} + +// resolveZfsPoolHistoryOnDelete resolves open alert history entries when a +// pool record is deleted (manually or because the pool disappeared), so the +// UI does not keep showing an ongoing alert for a pool that no longer exists. +func resolveZfsPoolHistoryOnDelete(e *core.RecordEvent) error { + resolveAllAlertHistoryRecords(e.App, e.Record.Id) + return e.Next() +} + +// shouldSendZfsPoolAlert reports whether a health transition warrants an alert. +// First observations of unhealthy pools and worsening transitions are reported. +func shouldSendZfsPoolAlert(oldSeverity, newSeverity int) bool { + return newSeverity > 1 && (oldSeverity == 0 || newSeverity > oldSeverity) +} + +// zfsPoolSeverity ranks pool health states: healthy (1), degraded (2), +// failed/unavailable (3), unknown (0). +func zfsPoolSeverity(health string) int { + switch health { + case "ONLINE": + return 1 + case "DEGRADED": + return 2 + case "FAULTED", "OFFLINE", "UNAVAIL", "REMOVED", "SUSPENDED": + return 3 + default: + return 0 + } +} + +// createZfsPoolHistoryRecord logs a pool health alert in the alerts history so +// it is visible in the UI without creating an editable alert configuration. +func createZfsPoolHistoryRecord(app core.App, userID, systemID, alertID, poolName string) error { + collection, err := app.FindCachedCollectionByNameOrId("alerts_history") + if err != nil { + return err + } + record := core.NewRecord(collection) + record.Set("user", userID) + record.Set("system", systemID) + record.Set("alert_id", alertID) + record.Set("name", "ZFS Pool: "+poolName) + return app.Save(record) +} + +// resolveAllAlertHistoryRecords resolves every open history entry for an alert +// record id (one per system user). +func resolveAllAlertHistoryRecords(app core.App, alertID string) { + records, err := app.FindRecordsByFilter( + "alerts_history", + "alert_id={:alert_id} && resolved=null", + "", 0, 0, + dbx.Params{"alert_id": alertID}, + ) + if err != nil || len(records) == 0 { + return + } + now := time.Now().UTC() + for _, record := range records { + record.Set("resolved", now) + if err := app.Save(record); err != nil { + app.Logger().Error("Failed to resolve ZFS alert history", "err", err, "recordId", record.Id) + } + } +} diff --git a/internal/alerts/alerts_zfs_disk_test.go b/internal/alerts/alerts_zfs_disk_test.go new file mode 100644 index 00000000..bd20c915 --- /dev/null +++ b/internal/alerts/alerts_zfs_disk_test.go @@ -0,0 +1,145 @@ +//go:build testing + +package alerts_test + +import ( + "encoding/json" + "testing" + "time" + + "github.com/henrygd/beszel/internal/entities/system" + beszelTests "github.com/henrygd/beszel/internal/tests" + + "github.com/pocketbase/dbx" + "github.com/pocketbase/pocketbase/tools/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestDiskAlertZfsPoolMultiMinute verifies that ZFS pool usage participates in +// the Disk threshold alert using historical per-minute values, mirroring the +// extra-filesystem behavior. +func TestDiskAlertZfsPoolMultiMinute(t *testing.T) { + hub, user := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + systems, err := beszelTests.CreateSystems(hub, 1, user.Id, "up") + require.NoError(t, err) + systemRecord := systems[0] + + diskAlert, err := beszelTests.CreateRecord(hub, "alerts", map[string]any{ + "name": "Disk", + "system": systemRecord.Id, + "user": user.Id, + "value": 80, // threshold: 80% + "min": 2, // requires historical averaging + }) + require.NoError(t, err) + + am := hub.GetAlertManager() + now := time.Now().UTC() + + poolHigh := map[string]*system.ZfsPool{ + "tank": {Total: 1000, Used: 920}, // 92% - above threshold + } + + recordTimes := []time.Duration{ + -180 * time.Second, + -90 * time.Second, + -60 * time.Second, + -30 * time.Second, + } + + for _, offset := range recordTimes { + stats := system.Stats{ + DiskPct: 30, // root disk at 30% - below threshold + ZfsPools: poolHigh, + } + statsJSON, _ := json.Marshal(stats) + + recordTime := now.Add(offset) + record, err := beszelTests.CreateRecord(hub, "system_stats", map[string]any{ + "system": systemRecord.Id, + "type": "1m", + "stats": string(statsJSON), + }) + require.NoError(t, err) + record.SetRaw("created", recordTime.Format(types.DefaultDateLayout)) + err = hub.SaveNoValidate(record) + require.NoError(t, err) + } + + combinedDataHigh := &system.CombinedData{ + Stats: system.Stats{ + DiskPct: 30, + ZfsPools: poolHigh, + }, + Info: system.Info{ + DiskPct: 30, + }, + } + + systemRecord.Set("updated", now) + err = hub.SaveNoValidate(systemRecord) + require.NoError(t, err) + + err = am.HandleSystemAlerts(systemRecord, combinedDataHigh) + require.NoError(t, err) + + time.Sleep(20 * time.Millisecond) + + diskAlert, err = hub.FindFirstRecordByFilter("alerts", "id={:id}", dbx.Params{"id": diskAlert.Id}) + require.NoError(t, err) + assert.True(t, diskAlert.GetBool("triggered"), + "Alert should be triggered when ZFS pool average (92%%) exceeds threshold (80%%)") + + // --- Resolution: pool drops to 50%, alert should resolve --- + + poolLow := map[string]*system.ZfsPool{ + "tank": {Total: 1000, Used: 500}, // 50% - below threshold + } + + newNow := now.Add(2 * time.Minute) + for _, offset := range recordTimes { + stats := system.Stats{ + DiskPct: 30, + ZfsPools: poolLow, + } + statsJSON, _ := json.Marshal(stats) + + recordTime := newNow.Add(offset) + record, err := beszelTests.CreateRecord(hub, "system_stats", map[string]any{ + "system": systemRecord.Id, + "type": "1m", + "stats": string(statsJSON), + }) + require.NoError(t, err) + record.SetRaw("created", recordTime.Format(types.DefaultDateLayout)) + err = hub.SaveNoValidate(record) + require.NoError(t, err) + } + + combinedDataLow := &system.CombinedData{ + Stats: system.Stats{ + DiskPct: 30, + ZfsPools: poolLow, + }, + Info: system.Info{ + DiskPct: 30, + }, + } + + systemRecord.Set("updated", newNow) + err = hub.SaveNoValidate(systemRecord) + require.NoError(t, err) + + err = am.HandleSystemAlerts(systemRecord, combinedDataLow) + require.NoError(t, err) + + time.Sleep(20 * time.Millisecond) + + diskAlert, err = hub.FindFirstRecordByFilter("alerts", "id={:id}", dbx.Params{"id": diskAlert.Id}) + require.NoError(t, err) + assert.False(t, diskAlert.GetBool("triggered"), + "Alert should be resolved when ZFS pool average (50%%) drops below threshold (80%%)") +} diff --git a/internal/alerts/alerts_zfs_key_test.go b/internal/alerts/alerts_zfs_key_test.go new file mode 100644 index 00000000..52e83c5b --- /dev/null +++ b/internal/alerts/alerts_zfs_key_test.go @@ -0,0 +1,15 @@ +//go:build testing + +package alerts + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestZfsDiskAlertKeyIsNamespaced(t *testing.T) { + assert.Equal(t, "zfs:tank", zfsDiskAlertKey("tank")) + assert.Equal(t, "Usage of ZFS pool tank", diskAlertDescriptor(zfsDiskAlertKey("tank"))) + assert.Equal(t, "Usage of tank", diskAlertDescriptor("tank")) +} diff --git a/internal/alerts/alerts_zfs_test.go b/internal/alerts/alerts_zfs_test.go new file mode 100644 index 00000000..708653d9 --- /dev/null +++ b/internal/alerts/alerts_zfs_test.go @@ -0,0 +1,292 @@ +//go:build testing + +package alerts_test + +import ( + "testing" + "time" + + beszelTests "github.com/henrygd/beszel/internal/tests" + "github.com/pocketbase/pocketbase/core" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestZfsPoolAlertOnlineToDegraded(t *testing.T) { + hub, user := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + system, err := beszelTests.CreateRecord(hub, "systems", map[string]any{ + "name": "test-system", + "users": []string{user.Id}, + "host": "127.0.0.1", + }) + assert.NoError(t, err) + + pool, err := beszelTests.CreateRecord(hub, "zfs_pools", map[string]any{ + "system": system.Id, + "name": "tank", + "health": "ONLINE", + }) + assert.NoError(t, err) + + // Re-fetch so PocketBase tracks original values + pool, err = hub.FindRecordById("zfs_pools", pool.Id) + assert.NoError(t, err) + + pool.Set("health", "DEGRADED") + err = hub.Save(pool) + assert.NoError(t, err) + + time.Sleep(50 * time.Millisecond) + + assert.EqualValues(t, 1, hub.TestMailer.TotalSend(), "should have 1 email sent after pool became DEGRADED") + lastMessage := hub.TestMailer.LastMessage() + assert.Contains(t, lastMessage.Subject, "ZFS pool DEGRADED on test-system") + assert.Contains(t, lastMessage.Subject, "tank") + assert.Contains(t, lastMessage.Text, "ONLINE to DEGRADED") +} + +func TestZfsPoolAlertDegradedToFaulted(t *testing.T) { + hub, user := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + system, err := beszelTests.CreateRecord(hub, "systems", map[string]any{ + "name": "test-system", + "users": []string{user.Id}, + "host": "127.0.0.1", + }) + assert.NoError(t, err) + + pool, err := beszelTests.CreateRecord(hub, "zfs_pools", map[string]any{ + "system": system.Id, + "name": "rpool", + "health": "DEGRADED", + }) + assert.NoError(t, err) + + pool, err = hub.FindRecordById("zfs_pools", pool.Id) + assert.NoError(t, err) + + pool.Set("health", "FAULTED") + err = hub.Save(pool) + assert.NoError(t, err) + + time.Sleep(50 * time.Millisecond) + + assert.EqualValues(t, 2, hub.TestMailer.TotalSend(), "should alert on initial DEGRADED state and later FAULTED transition") + lastMessage := hub.TestMailer.LastMessage() + assert.Contains(t, lastMessage.Subject, "ZFS pool FAULTED on test-system") +} + +func TestZfsPoolAlertNoAlertOnRecovery(t *testing.T) { + hub, user := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + system, err := beszelTests.CreateRecord(hub, "systems", map[string]any{ + "name": "test-system", + "users": []string{user.Id}, + "host": "127.0.0.1", + }) + assert.NoError(t, err) + + pool, err := beszelTests.CreateRecord(hub, "zfs_pools", map[string]any{ + "system": system.Id, + "name": "tank", + "health": "DEGRADED", + }) + assert.NoError(t, err) + + // Trigger a worsening alert first + pool, err = hub.FindRecordById("zfs_pools", pool.Id) + assert.NoError(t, err) + pool.Set("health", "FAULTED") + err = hub.Save(pool) + assert.NoError(t, err) + time.Sleep(50 * time.Millisecond) + assert.EqualValues(t, 2, hub.TestMailer.TotalSend(), "expected alerts for initial DEGRADED state and DEGRADED -> FAULTED") + + // Recovery back to ONLINE must not send a new alert + pool, err = hub.FindRecordById("zfs_pools", pool.Id) + assert.NoError(t, err) + pool.Set("health", "ONLINE") + err = hub.Save(pool) + assert.NoError(t, err) + time.Sleep(50 * time.Millisecond) + assert.EqualValues(t, 2, hub.TestMailer.TotalSend(), "recovery should not send a new alert") + + // And the open history entry should have been resolved + history, err := hub.FindRecordsByFilter("alerts_history", "alert_id={:alert_id}", "", 0, 0, map[string]any{"alert_id": pool.Id}) + assert.NoError(t, err) + requireHistoryResolved(t, history) +} + +func TestZfsPoolAlertUnknownHealthDoesNotResolve(t *testing.T) { + hub, user := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + system, err := beszelTests.CreateRecord(hub, "systems", map[string]any{ + "name": "test-system", + "users": []string{user.Id}, + "host": "127.0.0.1", + }) + require.NoError(t, err) + + pool, err := beszelTests.CreateRecord(hub, "zfs_pools", map[string]any{ + "system": system.Id, + "name": "tank", + "health": "DEGRADED", + }) + require.NoError(t, err) + time.Sleep(50 * time.Millisecond) + + pool, err = hub.FindRecordById("zfs_pools", pool.Id) + require.NoError(t, err) + pool.Set("health", "") + require.NoError(t, hub.Save(pool)) + time.Sleep(50 * time.Millisecond) + + history, err := hub.FindRecordsByFilter("alerts_history", "alert_id={:alert_id} && resolved=null", "", 0, 0, map[string]any{"alert_id": pool.Id}) + require.NoError(t, err) + require.Len(t, history, 1, "unknown health must not resolve an active alert") +} + +func TestZfsPoolAlertUnknownToFaulted(t *testing.T) { + hub, user := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + system, err := beszelTests.CreateRecord(hub, "systems", map[string]any{ + "name": "test-system", + "users": []string{user.Id}, + "host": "127.0.0.1", + }) + assert.NoError(t, err) + + pool, err := beszelTests.CreateRecord(hub, "zfs_pools", map[string]any{ + "system": system.Id, + "name": "tank", + "health": "", + }) + assert.NoError(t, err) + + pool, err = hub.FindRecordById("zfs_pools", pool.Id) + assert.NoError(t, err) + + pool.Set("health", "FAULTED") + err = hub.Save(pool) + assert.NoError(t, err) + + time.Sleep(50 * time.Millisecond) + + assert.EqualValues(t, 1, hub.TestMailer.TotalSend(), "should alert when a previously unknown pool becomes FAULTED") +} + +func TestZfsPoolAlertOnInitialUnhealthyState(t *testing.T) { + hub, user := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + system, err := beszelTests.CreateRecord(hub, "systems", map[string]any{ + "name": "test-system", + "users": []string{user.Id}, + "host": "127.0.0.1", + }) + require.NoError(t, err) + + pool, err := beszelTests.CreateRecord(hub, "zfs_pools", map[string]any{ + "system": system.Id, + "name": "tank", + "health": "DEGRADED", + }) + require.NoError(t, err) + time.Sleep(50 * time.Millisecond) + + require.EqualValues(t, 1, hub.TestMailer.TotalSend()) + assert.Contains(t, hub.TestMailer.LastMessage().Text, "first observed as DEGRADED") + + pool, err = hub.FindRecordById("zfs_pools", pool.Id) + require.NoError(t, err) + require.NoError(t, hub.Save(pool)) + time.Sleep(50 * time.Millisecond) + assert.EqualValues(t, 1, hub.TestMailer.TotalSend(), "unchanged unhealthy health must not duplicate alerts") +} + +func TestZfsPoolAlertWritesHistory(t *testing.T) { + hub, user := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + system, err := beszelTests.CreateRecord(hub, "systems", map[string]any{ + "name": "test-system", + "users": []string{user.Id}, + "host": "127.0.0.1", + }) + assert.NoError(t, err) + + pool, err := beszelTests.CreateRecord(hub, "zfs_pools", map[string]any{ + "system": system.Id, + "name": "tank", + "health": "ONLINE", + }) + assert.NoError(t, err) + + pool, err = hub.FindRecordById("zfs_pools", pool.Id) + assert.NoError(t, err) + pool.Set("health", "FAULTED") + err = hub.Save(pool) + assert.NoError(t, err) + + time.Sleep(50 * time.Millisecond) + + history, err := hub.FindRecordsByFilter("alerts_history", "alert_id={:alert_id}", "", 0, 0, map[string]any{"alert_id": pool.Id}) + assert.NoError(t, err) + require.Len(t, history, 1, "expected one history entry per user") + assert.Equal(t, "ZFS Pool: tank", history[0].GetString("name")) + assert.Equal(t, system.Id, history[0].GetString("system")) +} + +func TestZfsPoolAlertResolvedOnRecordDelete(t *testing.T) { + hub, user := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + system, err := beszelTests.CreateRecord(hub, "systems", map[string]any{ + "name": "test-system", + "users": []string{user.Id}, + "host": "127.0.0.1", + }) + assert.NoError(t, err) + + pool, err := beszelTests.CreateRecord(hub, "zfs_pools", map[string]any{ + "system": system.Id, + "name": "tank", + "health": "ONLINE", + }) + assert.NoError(t, err) + + // Trigger an alert so an open history entry exists. + pool, err = hub.FindRecordById("zfs_pools", pool.Id) + assert.NoError(t, err) + pool.Set("health", "FAULTED") + err = hub.Save(pool) + assert.NoError(t, err) + time.Sleep(50 * time.Millisecond) + + history, err := hub.FindRecordsByFilter("alerts_history", "alert_id={:alert_id} && resolved=null", "", 0, 0, map[string]any{"alert_id": pool.Id}) + assert.NoError(t, err) + require.Len(t, history, 1, "expected one open history entry") + + // Deleting the pool record must resolve the open entry. + err = hub.Delete(pool) + assert.NoError(t, err) + time.Sleep(50 * time.Millisecond) + + history, err = hub.FindRecordsByFilter("alerts_history", "alert_id={:alert_id}", "", 0, 0, map[string]any{"alert_id": pool.Id}) + assert.NoError(t, err) + require.Len(t, history, 1) + requireHistoryResolved(t, history) +} + +func requireHistoryResolved(t *testing.T, history []*core.Record) { + t.Helper() + for _, record := range history { + assert.False(t, record.GetDateTime("resolved").Time().IsZero(), "expected history entry to be resolved") + } +} diff --git a/internal/common/common-ws.go b/internal/common/common-ws.go index dee77918..3f69ba88 100644 --- a/internal/common/common-ws.go +++ b/internal/common/common-ws.go @@ -22,6 +22,8 @@ const ( GetSmartData // Request detailed systemd service info from agent GetSystemdInfo + // Request ZFS detail data from agent + GetZfsData // Add new actions here... ) @@ -64,6 +66,10 @@ type DataRequestOptions struct { IncludeDetails bool `cbor:"1,keyasint"` } +type ZfsDataRequest struct { + Force bool `cbor:"0,keyasint,omitempty"` +} + type ContainerLogsRequest struct { ContainerID string `cbor:"0,keyasint"` } diff --git a/internal/dockerfile_agent_alpine b/internal/dockerfile_agent_alpine index 09a20fe9..10bab26f 100644 --- a/internal/dockerfile_agent_alpine +++ b/internal/dockerfile_agent_alpine @@ -23,7 +23,7 @@ COPY --from=builder /agent /agent # AMD GPU name lookup (used by agent on Linux when /usr/share/libdrm/amdgpu.ids is read) COPY --from=builder /app/agent/test-data/amdgpu.ids /usr/share/libdrm/amdgpu.ids -RUN apk add --no-cache smartmontools +RUN apk add --no-cache smartmontools zfs # Ensure data persistence across container recreations VOLUME ["/var/lib/beszel-agent"] diff --git a/internal/entities/system/system.go b/internal/entities/system/system.go index a73db34e..66858397 100644 --- a/internal/entities/system/system.go +++ b/internal/entities/system/system.go @@ -33,8 +33,6 @@ type Stats struct { MaxNetworkSent float64 `json:"nsm,omitempty" cbor:"-"` MaxNetworkRecv float64 `json:"nrm,omitempty" cbor:"-"` Temperatures map[string]float64 `json:"t,omitempty" cbor:"20,keyasint,omitempty"` - Fans map[string]uint16 `json:"f,omitempty" cbor:"36,keyasint,omitempty"` - Batteries map[string]uint8 `json:"bats,omitempty" cbor:"37,keyasint,omitempty"` ExtraFs map[string]*FsStats `json:"efs,omitempty" cbor:"21,keyasint,omitempty"` GPUData map[string]GPUData `json:"g,omitempty" cbor:"22,keyasint,omitempty"` // LoadAvg1 float64 `json:"l1,omitempty" cbor:"23,keyasint,omitempty"` @@ -52,7 +50,20 @@ type Stats struct { CpuCoresUsage Uint8Slice `json:"cpus,omitempty" cbor:"34,keyasint,omitempty"` // per-core busy usage [CPU0..] DiskIoStats [6]float64 `json:"dios,omitzero" cbor:"35,keyasint,omitzero"` // [read time %, write time %, io utilization %, r_await ms, w_await ms, weighted io %] MaxDiskIoStats [6]float64 `json:"diosm,omitzero" cbor:"-"` // max values for DiskIoStats - DiskIOTotal [2]uint64 `json:"diot,omitzero" cbor:"38,keyasint,omitzero"` // [total read bytes, total write bytes] cumulative device counters + Fans map[string]uint16 `json:"f,omitempty" cbor:"36,keyasint,omitempty"` + Batteries map[string]uint8 `json:"bats,omitempty" cbor:"37,keyasint,omitempty"` + ZfsPools map[string]*ZfsPool `json:"z,omitempty" cbor:"39,keyasint,omitempty"` // ZFS pool metrics, keyed by pool name + DiskIOTotal [2]uint64 `json:"diot,omitzero" cbor:"38,keyasint,omitzero"` // [total read bytes, total write bytes] cumulative device counters + +} + +// ZfsPool holds per-pool ZFS metrics for a single collection interval. +type ZfsPool struct { + Total float64 `json:"d" cbor:"0,keyasint"` // total capacity in GiB + Used float64 `json:"du" cbor:"1,keyasint"` // allocated in GiB + ReadBytes uint64 `json:"rb,omitzero" cbor:"2,keyasint,omitzero"` // read throughput in bytes/s + WriteBytes uint64 `json:"wb,omitzero" cbor:"3,keyasint,omitzero"` // write throughput in bytes/s + Health string `json:"h,omitempty" cbor:"4,keyasint,omitempty"` // ONLINE, DEGRADED, FAULTED, ... } // Uint8Slice wraps []uint8 to customize JSON encoding while keeping CBOR efficient. @@ -183,6 +194,7 @@ type Details struct { Podman bool `cbor:"8,keyasint,omitempty"` MemoryTotal uint64 `cbor:"9,keyasint"` SmartInterval time.Duration `cbor:"10,keyasint,omitempty"` + ZfsInterval time.Duration `cbor:"11,keyasint,omitempty"` // interval for ZFS detail refresh } // Final data structure to return to the hub diff --git a/internal/entities/zfs/zfs.go b/internal/entities/zfs/zfs.go new file mode 100644 index 00000000..51d01723 --- /dev/null +++ b/internal/entities/zfs/zfs.go @@ -0,0 +1,45 @@ +// Package zfs defines the ZFS detail data exchanged between agent and hub. +package zfs + +// ZfsData is the detail payload returned by the agent for the GetZfsData action. +type ZfsData struct { + Pools []*PoolDetail `json:"pools,omitempty"` + Complete bool `json:"complete,omitempty"` +} + +// PoolDetail holds the verbose state of a single pool: capacity, health, +// scrub, vdev, and dataset information. +type PoolDetail struct { + Name string `json:"name"` + Health string `json:"health,omitempty"` + Size uint64 `json:"size,omitempty"` // bytes + Alloc uint64 `json:"alloc,omitempty"` // bytes + Free uint64 `json:"free,omitempty"` // bytes + Scrub *Scrub `json:"scrub,omitempty"` + Vdevs []*Vdev `json:"vdevs,omitempty"` + Datasets []*Dataset `json:"datasets,omitempty"` +} + +// Scrub holds the scrub (or resilver) status of a pool. +type Scrub struct { + State string `json:"state,omitempty"` // NONE, SCANNING, FINISHED, CANCELED + Progress string `json:"progress,omitempty"` + Errors uint64 `json:"errors,omitempty"` +} + +// Vdev is a single vdev (mirror, raidz, or leaf disk) with error counters. +type Vdev struct { + Name string `json:"name"` + State string `json:"state,omitempty"` + ReadErrs uint64 `json:"readErrs,omitempty"` + WriteErrs uint64 `json:"writeErrs,omitempty"` + ChecksumErrs uint64 `json:"checksumErrs,omitempty"` +} + +// Dataset is a single ZFS dataset with usage information. +type Dataset struct { + Name string `json:"name"` + Used uint64 `json:"used,omitempty"` + Avail uint64 `json:"avail,omitempty"` + Mountpoint string `json:"mount,omitempty"` +} diff --git a/internal/hub/api.go b/internal/hub/api.go index d2d009c0..c57d8dc8 100644 --- a/internal/hub/api.go +++ b/internal/hub/api.go @@ -125,6 +125,8 @@ func (h *Hub) registerApiRoutes(se *core.ServeEvent) error { apiAuth.DELETE("/user-alerts", alerts.DeleteUserAlerts) // refresh SMART devices for a system apiAuth.POST("/smart/refresh", h.refreshSmartData).BindFunc(excludeReadOnlyRole) + // refresh ZFS pool details for a system + apiAuth.POST("/zfs/refresh", h.refreshZfsData).BindFunc(excludeReadOnlyRole) // get systemd service details apiAuth.GET("/systemd/info", h.getSystemdInfo) // /containers routes @@ -389,3 +391,23 @@ func (h *Hub) refreshSmartData(e *core.RequestEvent) error { return e.JSON(http.StatusOK, map[string]string{"status": "ok"}) } + +// refreshZfsData handles POST /api/beszel/zfs/refresh requests +// Fetches fresh ZFS detail data from the agent and updates the collection +func (h *Hub) refreshZfsData(e *core.RequestEvent) error { + systemID := e.Request.URL.Query().Get("system") + if systemID == "" { + return e.BadRequestError("Invalid system parameter", nil) + } + + system, err := h.sm.GetSystem(systemID) + if err != nil || !system.HasUser(e.App, e.Auth) { + return e.NotFoundError("", nil) + } + + if err := system.FetchAndSaveZfsPools(true); err != nil { + return e.InternalServerError("", err) + } + + return e.JSON(http.StatusOK, map[string]string{"status": "ok"}) +} diff --git a/internal/hub/collections.go b/internal/hub/collections.go index 5b939a44..eb88da62 100644 --- a/internal/hub/collections.go +++ b/internal/hub/collections.go @@ -91,6 +91,12 @@ func setCollectionAuthSettings(app core.App) error { }); err != nil { return err } + if err := applyCollectionRules(app, []string{"zfs_pools"}, collectionRules{ + list: &systemScopedReadRule, + view: &systemScopedReadRule, + }); err != nil { + return err + } if err := applyCollectionRules(app, []string{"fingerprints"}, collectionRules{ list: &systemScopedReadRule, diff --git a/internal/hub/systems/system.go b/internal/hub/systems/system.go index 6c0879a1..43955b52 100644 --- a/internal/hub/systems/system.go +++ b/internal/hub/systems/system.go @@ -21,6 +21,7 @@ import ( "github.com/henrygd/beszel/internal/entities/smart" "github.com/henrygd/beszel/internal/entities/system" "github.com/henrygd/beszel/internal/entities/systemd" + "github.com/henrygd/beszel/internal/entities/zfs" "github.com/henrygd/beszel" @@ -49,6 +50,8 @@ type System struct { detailsFetched atomic.Bool // True if static system details have been fetched and saved smartFetching atomic.Bool // True if SMART devices are currently being fetched smartInterval time.Duration // Interval for periodic SMART data updates + zfsFetching atomic.Bool // True if ZFS pools are currently being fetched + zfsInterval time.Duration // Interval for periodic ZFS detail data updates } func (sm *SystemManager) NewSystem(systemId string) *System { @@ -154,6 +157,12 @@ func (sys *System) update() error { // to prevent premature expiration leading to new fetch if interval is different. sys.manager.smartFetchMap.UpdateExpiration(sys.Id, sys.smartInterval+time.Minute) } + // update zfs interval if it's set on the agent side + if data.Details.ZfsInterval > 0 { + sys.zfsInterval = data.Details.ZfsInterval + sys.manager.hub.Logger().Info("ZFS interval updated from agent details", "system", sys.Id, "interval", sys.zfsInterval.String()) + sys.manager.zfsFetchMap.UpdateExpiration(sys.Id, sys.zfsInterval+time.Minute) + } } // Fetch and save SMART devices when system first comes online or at intervals @@ -170,6 +179,20 @@ func (sys *System) update() error { } } + // Fetch and save ZFS pool details when system first comes online or at intervals + if backgroundZfsFetchEnabled() && sys.detailsFetched.Load() && sys.supportsZfsData() { + if sys.zfsInterval <= 0 { + sys.zfsInterval = time.Hour + } + if sys.shouldFetchZfs() && sys.zfsFetching.CompareAndSwap(false, true) { + sys.manager.hub.Logger().Info("ZFS fetch", "system", sys.Id, "interval", sys.zfsInterval.String()) + go func() { + defer sys.zfsFetching.Store(false) + _ = sys.FetchAndSaveZfsPools(false) + }() + } + } + return err } @@ -241,6 +264,10 @@ func (sys *System) createRecords(data *system.CombinedData) (*core.Record, error } } + if err := sys.syncZfsPoolHealth(txApp, data.Stats.ZfsPools); err != nil { + return err + } + // update system record (do this last because it triggers alerts and we need above records to be inserted first) systemRecord.Set("status", up) systemRecord.Set("info", data.Info) @@ -558,6 +585,15 @@ func (sys *System) FetchSmartDataFromAgent() (smart.SmartDataResponse, error) { return result, err } +// FetchZfsDataFromAgent fetches ZFS detail data from the agent. +func (sys *System) FetchZfsDataFromAgent(force bool) (*zfs.ZfsData, error) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + var result zfs.ZfsData + err := sys.request(ctx, common.GetZfsData, common.ZfsDataRequest{Force: force}, &result) + return &result, err +} + func makeStableHashId(strings ...string) string { hash := fnv.New32a() for _, str := range strings { @@ -728,6 +764,7 @@ func (s *System) createSSHClient() error { } s.agentVersion, _ = extractAgentVersion(string(client.Conn.ServerVersion())) s.manager.resetFailedSmartFetchState(s.Id) + s.manager.resetFailedZfsFetchState(s.Id) return nil } diff --git a/internal/hub/systems/system_manager.go b/internal/hub/systems/system_manager.go index 42f9562d..8be62379 100644 --- a/internal/hub/systems/system_manager.go +++ b/internal/hub/systems/system_manager.go @@ -46,6 +46,7 @@ type SystemManager struct { systems *store.Store[string, *System] // Thread-safe store of active systems sshConfig *ssh.ClientConfig // SSH client configuration for system connections smartFetchMap *expirymap.ExpiryMap[smartFetchState] // Stores last SMART fetch time/result; TTL is only for cleanup + zfsFetchMap *expirymap.ExpiryMap[zfsFetchState] // Stores last ZFS fetch time/result; TTL is only for cleanup ctx context.Context // Cancelled when the app terminates cancel context.CancelFunc // Cancels ctx and all child system contexts } @@ -67,6 +68,7 @@ func NewSystemManager(hub hubLike) *SystemManager { systems: store.New(map[string]*System{}), hub: hub, smartFetchMap: expirymap.New[smartFetchState](time.Hour), + zfsFetchMap: expirymap.New[zfsFetchState](time.Hour), } sm.ctx, sm.cancel = context.WithCancel(context.Background()) return sm @@ -343,6 +345,15 @@ func (sm *SystemManager) resetFailedSmartFetchState(systemID string) { } } +// resetFailedZfsFetchState clears only failed ZFS cooldown entries so a fresh +// agent reconnect retries ZFS discovery immediately after configuration changes. +func (sm *SystemManager) resetFailedZfsFetchState(systemID string) { + state, ok := sm.zfsFetchMap.GetOk(systemID) + if ok && !state.Successful { + sm.zfsFetchMap.Remove(systemID) + } +} + // createSSHClientConfig initializes the SSH client configuration for connecting to an agent's server func (sm *SystemManager) createSSHClientConfig() error { privateKey, err := sm.hub.GetSSHKey("") diff --git a/internal/hub/systems/system_zfs.go b/internal/hub/systems/system_zfs.go new file mode 100644 index 00000000..221160f2 --- /dev/null +++ b/internal/hub/systems/system_zfs.go @@ -0,0 +1,193 @@ +package systems + +import ( + "database/sql" + "errors" + "fmt" + "time" + + "github.com/henrygd/beszel" + "github.com/henrygd/beszel/internal/entities/system" + "github.com/henrygd/beszel/internal/entities/zfs" + "github.com/pocketbase/dbx" + "github.com/pocketbase/pocketbase/core" +) + +var errIncompleteZfsData = errors.New("incomplete ZFS pool inventory") + +type zfsFetchState struct { + LastAttempt int64 + Successful bool +} + +func (sys *System) supportsZfsData() bool { + return sys.agentVersion.GTE(beszel.MinVersionZfsData) +} + +// FetchAndSaveZfsPools fetches ZFS detail data from the agent and saves it to +// the database. force bypasses the agent's detail cache for manual refreshes. +func (sys *System) FetchAndSaveZfsPools(force bool) error { + zfsData, err := sys.FetchZfsDataFromAgent(force) + if err != nil { + sys.recordZfsFetchResult(err, 0) + return err + } + if zfsData == nil || !zfsData.Complete { + err = errIncompleteZfsData + sys.recordZfsFetchResult(err, 0) + return err + } + err = sys.saveZfsPools(zfsData) + sys.recordZfsFetchResult(err, len(zfsData.Pools)) + return err +} + +// recordZfsFetchResult stores a cooldown entry for the ZFS interval and marks +// whether the last fetch produced any pools, so failed setup can retry on reconnect. +func (sys *System) recordZfsFetchResult(err error, poolCount int) { + if sys.manager == nil { + return + } + interval := sys.zfsFetchInterval() + success := err == nil && poolCount > 0 + if sys.manager.hub != nil { + sys.manager.hub.Logger().Info("ZFS fetch result", "system", sys.Id, "success", success, "pools", poolCount, "interval", interval.String(), "err", err) + } + sys.manager.zfsFetchMap.Set(sys.Id, zfsFetchState{LastAttempt: time.Now().UnixMilli(), Successful: success}, interval+time.Minute) +} + +// shouldFetchZfs returns true when there is no active ZFS cooldown entry for this system. +func (sys *System) shouldFetchZfs() bool { + if sys.manager == nil { + return true + } + state, ok := sys.manager.zfsFetchMap.GetOk(sys.Id) + if !ok { + return true + } + return !time.UnixMilli(state.LastAttempt).Add(sys.zfsFetchInterval()).After(time.Now()) +} + +// zfsFetchInterval returns the agent-provided ZFS interval or the default when unset. +func (sys *System) zfsFetchInterval() time.Duration { + if sys.zfsInterval > 0 { + return sys.zfsInterval + } + return time.Hour +} + +// saveZfsPools saves ZFS pool detail data to the zfs_pools collection and +// removes records for pools no longer reported by a complete agent inventory. +func (sys *System) saveZfsPools(zfsData *zfs.ZfsData) error { + if zfsData == nil || !zfsData.Complete { + return errIncompleteZfsData + } + + hub := sys.manager.hub + collection, err := hub.FindCachedCollectionByNameOrId("zfs_pools") + if err != nil { + return err + } + + return hub.RunInTransaction(func(txApp core.App) error { + alive := make(map[string]bool, len(zfsData.Pools)) + for _, pool := range zfsData.Pools { + if pool == nil { + continue + } + alive[pool.Name] = true + if err := sys.upsertZfsPoolRecord(txApp, collection, pool); err != nil { + return err + } + } + + existing, err := txApp.FindRecordsByFilter( + collection, + "system={:system}", + "", 0, 0, + dbx.Params{"system": sys.Id}, + ) + if err != nil { + return err + } + for _, record := range existing { + if !alive[record.GetString("name")] { + if err := txApp.Delete(record); err != nil { + return err + } + } + } + return nil + }) +} + +func (sys *System) upsertZfsPoolRecord(app core.App, collection *core.Collection, pool *zfs.PoolDetail) error { + recordID := makeStableHashId(sys.Id, pool.Name) + + record, err := app.FindRecordById(collection, recordID) + if err != nil { + if !errors.Is(err, sql.ErrNoRows) { + return err + } + record = core.NewRecord(collection) + record.Set("id", recordID) + } + + record.Set("system", sys.Id) + record.Set("name", pool.Name) + record.Set("health", pool.Health) + record.Set("size", pool.Size) + record.Set("alloc", pool.Alloc) + record.Set("free", pool.Free) + record.Set("scrub", pool.Scrub) + record.Set("vdevs", pool.Vdevs) + record.Set("datasets", pool.Datasets) + record.Set("details_updated", time.Now().UTC()) + + return app.SaveNoValidate(record) +} + +// syncZfsPoolHealth persists newly discovered pools and health transitions from +// regular system samples. Detailed fields remain owned by the hourly refresh. +func (sys *System) syncZfsPoolHealth(app core.App, pools map[string]*system.ZfsPool) error { + if len(pools) == 0 { + return nil + } + collection, err := app.FindCachedCollectionByNameOrId("zfs_pools") + if err != nil { + return err + } + const gib = 1024 * 1024 * 1024 + for name, pool := range pools { + if pool == nil { + continue + } + recordID := makeStableHashId(sys.Id, name) + record, err := app.FindRecordById(collection, recordID) + if err != nil { + if !errors.Is(err, sql.ErrNoRows) { + return err + } + record = core.NewRecord(collection) + record.Set("id", recordID) + record.Set("system", sys.Id) + record.Set("name", name) + record.Set("health", pool.Health) + record.Set("size", uint64(pool.Total*gib)) + record.Set("alloc", uint64(pool.Used*gib)) + record.Set("free", uint64(max(pool.Total-pool.Used, 0)*gib)) + if err := app.SaveNoValidate(record); err != nil { + return fmt.Errorf("creating ZFS pool summary %q: %w", name, err) + } + continue + } + if record.GetString("health") == pool.Health { + continue + } + record.Set("health", pool.Health) + if err := app.SaveNoValidate(record); err != nil { + return fmt.Errorf("updating ZFS pool health %q: %w", name, err) + } + } + return nil +} diff --git a/internal/hub/systems/system_zfs_test.go b/internal/hub/systems/system_zfs_test.go new file mode 100644 index 00000000..dd4433d4 --- /dev/null +++ b/internal/hub/systems/system_zfs_test.go @@ -0,0 +1,153 @@ +//go:build testing + +package systems + +import ( + "errors" + "testing" + "time" + + "github.com/blang/semver" + "github.com/henrygd/beszel/internal/entities/system" + "github.com/henrygd/beszel/internal/entities/zfs" + "github.com/henrygd/beszel/internal/hub/expirymap" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSupportsZfsData(t *testing.T) { + sys := &System{agentVersion: semver.MustParse("0.18.8")} + assert.False(t, sys.supportsZfsData()) + + sys.agentVersion = semver.MustParse("0.18.9") + assert.True(t, sys.supportsZfsData()) +} + +func TestRecordZfsFetchResult(t *testing.T) { + sm := &SystemManager{zfsFetchMap: expirymap.New[zfsFetchState](time.Hour)} + t.Cleanup(sm.zfsFetchMap.StopCleaner) + + sys := &System{ + Id: "system-1", + manager: sm, + zfsInterval: time.Hour, + } + + // Successful fetch with pools + sys.recordZfsFetchResult(nil, 2) + state, ok := sm.zfsFetchMap.GetOk(sys.Id) + assert.True(t, ok, "expected zfs fetch result to be stored") + assert.True(t, state.Successful, "expected successful fetch state to be recorded") + + // Failed fetch + sys.recordZfsFetchResult(errors.New("failed"), 0) + state, ok = sm.zfsFetchMap.GetOk(sys.Id) + assert.True(t, ok, "expected failed zfs fetch state to be stored") + assert.False(t, state.Successful, "expected failed zfs fetch state to be marked unsuccessful") + + // Successful fetch but no pools + sys.recordZfsFetchResult(nil, 0) + state, ok = sm.zfsFetchMap.GetOk(sys.Id) + assert.True(t, ok, "expected fetch with zero pools to be stored") + assert.False(t, state.Successful, "expected fetch with zero pools to be marked unsuccessful") +} + +func TestShouldFetchZfs(t *testing.T) { + sm := &SystemManager{zfsFetchMap: expirymap.New[zfsFetchState](time.Hour)} + t.Cleanup(sm.zfsFetchMap.StopCleaner) + + sys := &System{ + Id: "system-1", + manager: sm, + zfsInterval: time.Hour, + } + + assert.True(t, sys.shouldFetchZfs(), "expected initial zfs fetch to be allowed") + + sys.recordZfsFetchResult(errors.New("failed"), 0) + assert.False(t, sys.shouldFetchZfs(), "expected zfs fetch to be blocked while interval entry exists") + + sm.zfsFetchMap.Remove(sys.Id) + assert.True(t, sys.shouldFetchZfs(), "expected zfs fetch to be allowed after interval entry is cleared") +} + +func TestZfsFetchIntervalDefault(t *testing.T) { + sys := &System{} + assert.Equal(t, time.Hour, sys.zfsFetchInterval()) + + sys.zfsInterval = 5 * time.Minute + assert.Equal(t, 5*time.Minute, sys.zfsFetchInterval()) +} + +func TestResetFailedZfsFetchState(t *testing.T) { + sm := &SystemManager{zfsFetchMap: expirymap.New[zfsFetchState](time.Hour)} + t.Cleanup(sm.zfsFetchMap.StopCleaner) + + sm.zfsFetchMap.Set("system-1", zfsFetchState{LastAttempt: time.Now().UnixMilli(), Successful: false}, time.Hour) + sm.resetFailedZfsFetchState("system-1") + _, ok := sm.zfsFetchMap.GetOk("system-1") + assert.False(t, ok, "expected failed zfs fetch state to be cleared on reconnect") + + sm.zfsFetchMap.Set("system-1", zfsFetchState{LastAttempt: time.Now().UnixMilli(), Successful: true}, time.Hour) + sm.resetFailedZfsFetchState("system-1") + _, ok = sm.zfsFetchMap.GetOk("system-1") + assert.True(t, ok, "expected successful zfs fetch state to be preserved") +} + +func TestSaveZfsPoolsCompleteEmptyPrunesFinalPool(t *testing.T) { + sys, app := newTestSystemWithHub(t) + require.NoError(t, sys.saveZfsPools(&zfs.ZfsData{ + Complete: true, + Pools: []*zfs.PoolDetail{{Name: "tank", Health: "ONLINE"}}, + })) + records, err := app.FindRecordsByFilter("zfs_pools", "system={:system}", "", 0, 0, map[string]any{"system": sys.Id}) + require.NoError(t, err) + require.Len(t, records, 1) + assert.False(t, records[0].GetDateTime("details_updated").Time().IsZero()) + + require.NoError(t, sys.saveZfsPools(&zfs.ZfsData{Complete: true})) + records, err = app.FindRecordsByFilter("zfs_pools", "system={:system}", "", 0, 0, map[string]any{"system": sys.Id}) + require.NoError(t, err) + assert.Empty(t, records) +} + +func TestSaveZfsPoolsIncompletePreservesRecords(t *testing.T) { + sys, app := newTestSystemWithHub(t) + require.NoError(t, sys.saveZfsPools(&zfs.ZfsData{ + Complete: true, + Pools: []*zfs.PoolDetail{{Name: "tank", Health: "ONLINE"}}, + })) + assert.ErrorIs(t, sys.saveZfsPools(&zfs.ZfsData{}), errIncompleteZfsData) + records, err := app.FindRecordsByFilter("zfs_pools", "system={:system}", "", 0, 0, map[string]any{"system": sys.Id}) + require.NoError(t, err) + assert.Len(t, records, 1) +} + +func TestSyncZfsPoolHealthWritesOnlyTransitions(t *testing.T) { + sys, app := newTestSystemWithHub(t) + collection, err := app.FindCachedCollectionByNameOrId("zfs_pools") + require.NoError(t, err) + + require.NoError(t, sys.syncZfsPoolHealth(app, map[string]*system.ZfsPool{ + "tank": {Total: 100, Used: 25, Health: "ONLINE"}, + })) + record, err := app.FindRecordById(collection, makeStableHashId(sys.Id, "tank")) + require.NoError(t, err) + firstUpdated := record.GetDateTime("updated") + assert.Equal(t, "ONLINE", record.GetString("health")) + assert.EqualValues(t, 100*1024*1024*1024, record.GetInt("size")) + + require.NoError(t, sys.syncZfsPoolHealth(app, map[string]*system.ZfsPool{ + "tank": {Total: 100, Used: 30, Health: "ONLINE"}, + })) + record, err = app.FindRecordById(collection, record.Id) + require.NoError(t, err) + assert.Equal(t, firstUpdated, record.GetDateTime("updated")) + + require.NoError(t, sys.syncZfsPoolHealth(app, map[string]*system.ZfsPool{ + "tank": {Total: 100, Used: 30, Health: "DEGRADED"}, + })) + record, err = app.FindRecordById(collection, record.Id) + require.NoError(t, err) + assert.Equal(t, "DEGRADED", record.GetString("health")) +} diff --git a/internal/hub/systems/systems_production.go b/internal/hub/systems/systems_production.go index 754c555b..b41dfa4d 100644 --- a/internal/hub/systems/systems_production.go +++ b/internal/hub/systems/systems_production.go @@ -7,3 +7,6 @@ package systems // The hub integration tests create/replace systems and clean up the test apps quickly. // Background SMART fetching can outlive teardown and crash in PocketBase internals (nil DB). func backgroundSmartFetchEnabled() bool { return true } + +// Background ZFS fetching follows the same policy as SMART fetching. +func backgroundZfsFetchEnabled() bool { return true } diff --git a/internal/hub/systems/systems_test_helpers.go b/internal/hub/systems/systems_test_helpers.go index b6ae4782..5d69ac12 100644 --- a/internal/hub/systems/systems_test_helpers.go +++ b/internal/hub/systems/systems_test_helpers.go @@ -17,6 +17,9 @@ import ( // the automatic background fetch during tests. func backgroundSmartFetchEnabled() bool { return false } +// Background ZFS fetching follows the same policy as SMART fetching. +func backgroundZfsFetchEnabled() bool { return false } + // TESTING ONLY: GetSystemCount returns the number of systems in the store func (sm *SystemManager) GetSystemCount() int { return sm.systems.Length() @@ -115,6 +118,7 @@ func (sm *SystemManager) RemoveAllSystems() { sm.RemoveSystem(system.Id) } sm.smartFetchMap.StopCleaner() + sm.zfsFetchMap.StopCleaner() } // ResetContextForTesting replaces the manager context for a new synctest bubble. diff --git a/internal/migrations/zfs_pools.go b/internal/migrations/zfs_pools.go new file mode 100644 index 00000000..a0c9c1bc --- /dev/null +++ b/internal/migrations/zfs_pools.go @@ -0,0 +1,184 @@ +package migrations + +import ( + "github.com/pocketbase/pocketbase/core" + m "github.com/pocketbase/pocketbase/migrations" +) + +// Creates the zfs_pools collection for per-system ZFS pool detail data +// (pool health, capacity, scrub, vdevs, datasets). Upserts rather than +// deletes missing collections, so it is safe on fresh and existing installs. +func init() { + m.Register(func(app core.App) error { + // update collections + jsonData := `[ + { + "createRule": null, + "deleteRule": null, + "fields": [ + { + "autogeneratePattern": "[a-z0-9]{15}", + "hidden": false, + "id": "text3208210256", + "max": 15, + "min": 15, + "name": "id", + "pattern": "^[a-z0-9]+$", + "presentable": false, + "primaryKey": true, + "required": true, + "system": true, + "type": "text" + }, + { + "cascadeDelete": true, + "collectionId": "2hz5ncl8tizk5nx", + "hidden": false, + "id": "relation1204987316", + "maxSelect": 1, + "minSelect": 0, + "name": "system", + "presentable": false, + "required": true, + "system": false, + "type": "relation" + }, + { + "autogeneratePattern": "", + "hidden": false, + "id": "text7739291048", + "max": 0, + "min": 0, + "name": "name", + "pattern": "", + "presentable": false, + "primaryKey": false, + "required": false, + "system": false, + "type": "text" + }, + { + "autogeneratePattern": "", + "hidden": false, + "id": "text5528164482", + "max": 0, + "min": 0, + "name": "health", + "pattern": "", + "presentable": false, + "primaryKey": false, + "required": false, + "system": false, + "type": "text" + }, + { + "hidden": false, + "id": "number8862034195", + "max": null, + "min": null, + "name": "size", + "onlyInt": true, + "presentable": false, + "required": false, + "system": false, + "type": "number" + }, + { + "hidden": false, + "id": "number4418907321", + "max": null, + "min": null, + "name": "alloc", + "onlyInt": true, + "presentable": false, + "required": false, + "system": false, + "type": "number" + }, + { + "hidden": false, + "id": "number2904183765", + "max": null, + "min": null, + "name": "free", + "onlyInt": true, + "presentable": false, + "required": false, + "system": false, + "type": "number" + }, + { + "hidden": false, + "id": "json4466109723", + "maxSize": 0, + "name": "scrub", + "presentable": false, + "required": false, + "system": false, + "type": "json" + }, + { + "hidden": false, + "id": "json9012873456", + "maxSize": 0, + "name": "vdevs", + "presentable": false, + "required": false, + "system": false, + "type": "json" + }, + { + "hidden": false, + "id": "json7182045639", + "maxSize": 0, + "name": "datasets", + "presentable": false, + "required": false, + "system": false, + "type": "json" + }, + { + "hidden": false, + "id": "date9274163058", + "max": "", + "min": "", + "name": "details_updated", + "presentable": false, + "required": false, + "system": false, + "type": "date" + }, + { + "hidden": false, + "id": "autodate3332085495", + "name": "updated", + "onCreate": true, + "onUpdate": true, + "presentable": false, + "system": false, + "type": "autodate" + } + ], + "id": "pbc_8441057391", + "indexes": [ + "CREATE INDEX ` + "`" + `idx_zfsPoolsSystem` + "`" + ` ON ` + "`" + `zfs_pools` + "`" + ` (` + "`" + `system` + "`" + `)" + ], + "listRule": null, + "name": "zfs_pools", + "system": false, + "type": "base", + "updateRule": null, + "viewRule": null + } +]` + + err := app.ImportCollectionsByMarshaledJSON([]byte(jsonData), false) + if err != nil { + return err + } + + return nil + }, func(app core.App) error { + return nil + }) +} diff --git a/internal/records/records.go b/internal/records/records.go index 465d0bf2..4b27e57a 100644 --- a/internal/records/records.go +++ b/internal/records/records.go @@ -127,6 +127,7 @@ func (rm *RecordManager) CreateLongerRecords() { "created": shorterRecordPeriod, }, )). + OrderBy("created"). All(&recordIds) // continue if not enough shorter records @@ -196,6 +197,7 @@ func AverageSystemStatsSlice(records []system.Stats) system.Stats { tempCount := float64(0) var fanSums map[string]uint64 fanCount := uint64(0) + zfsPoolCounts := make(map[string]uint64) // Accumulate totals for i := range records { @@ -336,6 +338,31 @@ func AverageSystemStatsSlice(records []system.Stats) system.Stats { } } + // Accumulate ZFS pool stats. Counts are tracked per entry so a pool + // missing from some samples is not averaged as zero. + if stats.ZfsPools != nil { + if sum.ZfsPools == nil { + sum.ZfsPools = make(map[string]*system.ZfsPool, len(stats.ZfsPools)) + } + for name, value := range stats.ZfsPools { + if value == nil { + continue + } + pool := sum.ZfsPools[name] + if pool == nil { + pool = &system.ZfsPool{} + sum.ZfsPools[name] = pool + } + pool.Total += value.Total + pool.Used += value.Used + pool.ReadBytes += value.ReadBytes + pool.WriteBytes += value.WriteBytes + if value.Health != "" { + pool.Health = value.Health + } + zfsPoolCounts[name]++ + } + } // Accumulate GPU data if stats.GPUData != nil { if sum.GPUData == nil { @@ -446,6 +473,14 @@ func AverageSystemStatsSlice(records []system.Stats) system.Stats { } } + // Average ZFS pool stats. + for name, pool := range sum.ZfsPools { + entryCount := zfsPoolCounts[name] + pool.Total = twoDecimals(pool.Total / float64(entryCount)) + pool.Used = twoDecimals(pool.Used / float64(entryCount)) + pool.ReadBytes /= entryCount + pool.WriteBytes /= entryCount + } // Average GPU data if sum.GPUData != nil { for id := range sum.GPUData { diff --git a/internal/records/records_averaging_test.go b/internal/records/records_averaging_test.go index bc974499..243b8aa3 100644 --- a/internal/records/records_averaging_test.go +++ b/internal/records/records_averaging_test.go @@ -669,6 +669,33 @@ func TestAverageSystemStatsSlice_MixedOptionalFields(t *testing.T) { assert.Equal(t, 20.0, result.GPUData["gpu0"].Usage) } +func TestAverageSystemStatsSlice_Zfs(t *testing.T) { + input := []system.Stats{ + { + ZfsPools: map[string]*system.ZfsPool{ + "tank": {Total: 100, Used: 40, ReadBytes: 100, WriteBytes: 200, Health: "ONLINE"}, + }, + }, + {}, + { + ZfsPools: map[string]*system.ZfsPool{ + "tank": {Total: 120, Used: 60, ReadBytes: 300, WriteBytes: 400, Health: "DEGRADED"}, + "backup": {Total: 50, Used: 10, ReadBytes: 25, WriteBytes: 50, Health: "ONLINE"}, + }, + }, + } + + result := records.AverageSystemStatsSlice(input) + + require.Len(t, result.ZfsPools, 2) + assert.Equal(t, &system.ZfsPool{ + Total: 110, Used: 50, ReadBytes: 200, WriteBytes: 300, Health: "DEGRADED", + }, result.ZfsPools["tank"]) + assert.Equal(t, &system.ZfsPool{ + Total: 50, Used: 10, ReadBytes: 25, WriteBytes: 50, Health: "ONLINE", + }, result.ZfsPools["backup"]) +} + // Tests with 10 records matching the common real-world case (10 x 1m -> 1 x 10m). func TestAverageSystemStatsSlice_TenRecords(t *testing.T) { input := make([]system.Stats, 10) diff --git a/internal/site/src/components/routes/system.tsx b/internal/site/src/components/routes/system.tsx index 495814ed..34a54cd6 100644 --- a/internal/site/src/components/routes/system.tsx +++ b/internal/site/src/components/routes/system.tsx @@ -8,10 +8,11 @@ import { useSystemData } from "./system/use-system-data" import { CpuChart, ContainerCpuChart } from "./system/charts/cpu-charts" import { MemoryChart, ContainerMemoryChart, SwapChart } from "./system/charts/memory-charts" import { RootDiskCharts, ExtraFsCharts } from "./system/charts/disk-charts" +import { ZfsCharts } from "./system/charts/zfs-charts" import { BandwidthChart, ContainerNetworkChart } from "./system/charts/network-charts" import { TemperatureChart, FanChart, BatteryChart } from "./system/charts/sensor-charts" import { GpuPowerChart, GpuDetailCharts } from "./system/charts/gpu-charts" -import { LazyContainersTable, LazySmartTable, LazySystemdTable } from "./system/lazy-tables" +import { LazyContainersTable, LazySmartTable, LazySystemdTable, LazyZfsTable } from "./system/lazy-tables" import { LoadAverageChart } from "./system/charts/load-average-chart" import { ContainerIcon, CpuIcon, HardDriveIcon, TerminalSquareIcon } from "lucide-react" import { GpuIcon } from "../ui/icons" @@ -63,6 +64,7 @@ export default memo(function SystemDetail({ id }: { id: string }) { const hasContainersTable = hasContainers && compareSemVer(chartData.agentVersion, SEMVER_0_14_0) >= 0 const hasSystemd = system.info.sv const hasGpu = hasGpuData || hasGpuPowerData + const hasZfs = Object.keys(systemStats.at(-1)?.stats?.z ?? {}).length > 0 // keep tabsRef in sync for keyboard navigation const tabs = ["core", "disk"] @@ -142,6 +144,10 @@ export default memo(function SystemDetail({ id }: { id: string }) { + {hasZfs && } + + {hasZfs && } + {maybeHasSmartData && } {hasContainersTable && } @@ -204,6 +210,8 @@ export default memo(function SystemDetail({ id }: { id: string }) { + {hasZfs && } + {hasZfs && } {maybeHasSmartData && } )} diff --git a/internal/site/src/components/routes/system/charts/zfs-charts.tsx b/internal/site/src/components/routes/system/charts/zfs-charts.tsx new file mode 100644 index 00000000..8bf141f8 --- /dev/null +++ b/internal/site/src/components/routes/system/charts/zfs-charts.tsx @@ -0,0 +1,130 @@ +import { t } from "@lingui/core/macro" +import AreaChartDefault from "@/components/charts/area-chart" +import { decimalString, formatBytes, toFixedFloat } from "@/lib/utils" +import type { SystemStatsRecord } from "@/types" +import { ChartCard } from "../chart-card" +import { Unit } from "@/lib/enums" +import { useStore } from "@nanostores/react" +import { $userSettings } from "@/lib/stores" +import type { SystemData } from "../use-system-data" + +// Accessors for ZFS metrics +const poolUsage = + (name: string) => + ({ stats }: SystemStatsRecord) => + stats?.z?.[name]?.du ?? 0 +const poolRead = + (name: string) => + ({ stats }: SystemStatsRecord) => + stats?.z?.[name]?.rb ?? 0 +const poolWrite = + (name: string) => + ({ stats }: SystemStatsRecord) => + stats?.z?.[name]?.wb ?? 0 + +export function ZfsPoolUsageChart({ systemData, poolName }: { systemData: SystemData; poolName: string }) { + const { chartData, grid, dataEmpty } = systemData + const latest = chartData.systemStats.at(-1)?.stats + const pool = latest?.z?.[poolName] + if (!pool) { + return null + } + let poolTotal = pool.d + // round to nearest GB + if (poolTotal >= 100) { + poolTotal = Math.round(poolTotal) + } + + return ( + + { + const { value, unit } = formatBytes(val * 1024, false, Unit.Bytes, true) + return `${toFixedFloat(value, value >= 10 ? 0 : 1)} ${unit}` + }} + contentFormatter={({ value }) => { + const { value: convertedValue, unit } = formatBytes(value * 1024, false, Unit.Bytes, true) + return `${decimalString(convertedValue, convertedValue >= 100 ? 1 : 2)} ${unit}` + }} + dataPoints={[ + { + label: t`Pool Usage`, + dataKey: poolUsage(poolName), + color: 4, + opacity: 0.4, + }, + ]} + /> + + ) +} + +export function ZfsPoolIOChart({ systemData, poolName }: { systemData: SystemData; poolName: string }) { + const { chartData, grid, dataEmpty } = systemData + const userSettings = useStore($userSettings) + if (!chartData.systemStats?.length) { + return null + } + return ( + + { + const { value, unit } = formatBytes(val, true, userSettings.unitDisk, false) + return `${toFixedFloat(value, value >= 10 ? 0 : 1)} ${unit}` + }} + contentFormatter={({ value }) => { + const { value: convertedValue, unit } = formatBytes(value, true, userSettings.unitDisk, false) + return `${decimalString(convertedValue, convertedValue >= 100 ? 1 : 2)} ${unit}` + }} + /> + + ) +} + +/** ZFS section: one stacked usage card per pool plus per-pool I/O cards. */ +export function ZfsCharts({ systemData }: { systemData: SystemData }) { + const latest = systemData.chartData.systemStats?.at(-1)?.stats + const pools = latest?.z ?? {} + if (Object.keys(pools).length === 0) { + return null + } + return ( +
+ {Object.keys(pools).map((poolName) => ( +
+ + +
+ ))} +
+ ) +} diff --git a/internal/site/src/components/routes/system/lazy-tables.tsx b/internal/site/src/components/routes/system/lazy-tables.tsx index f487369c..afe14e3c 100644 --- a/internal/site/src/components/routes/system/lazy-tables.tsx +++ b/internal/site/src/components/routes/system/lazy-tables.tsx @@ -24,6 +24,17 @@ export function LazySmartTable({ systemId }: { systemId: string }) { ) } +const ZfsTable = lazy(() => import("./zfs-table")) + +export function LazyZfsTable({ systemId }: { systemId: string }) { + const { isIntersecting, ref } = useIntersectionObserver({ rootMargin: "90px" }) + return ( +
+ {isIntersecting && } +
+ ) +} + const SystemdTable = lazy(() => import("../../systemd-table/systemd-table")) export function LazySystemdTable({ systemId }: { systemId: string }) { diff --git a/internal/site/src/components/routes/system/zfs-table.tsx b/internal/site/src/components/routes/system/zfs-table.tsx new file mode 100644 index 00000000..f83c9fe1 --- /dev/null +++ b/internal/site/src/components/routes/system/zfs-table.tsx @@ -0,0 +1,652 @@ +import { Alert, AlertDescription, AlertTitle } from "@/components/ui/alert" +import { Badge } from "@/components/ui/badge" +import { Button } from "@/components/ui/button" +import { Card, CardDescription, CardHeader, CardTitle } from "@/components/ui/card" +import { DropdownMenu, DropdownMenuContent, DropdownMenuItem, DropdownMenuTrigger } from "@/components/ui/dropdown-menu" +import { Input } from "@/components/ui/input" +import { Separator } from "@/components/ui/separator" +import { Sheet, SheetContent, SheetDescription, SheetHeader, SheetTitle } from "@/components/ui/sheet" +import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@/components/ui/table" +import { isReadOnlyUser, pb } from "@/lib/api" +import { cn, formatBytes, formatShortDate, hourWithSeconds, toFixedFloat } from "@/lib/utils" +import type { ZfsDataset, ZfsPoolRecord, ZfsVdev } from "@/types" +import { t } from "@lingui/core/macro" +import { Trans } from "@lingui/react/macro" +import type { Column, ColumnDef } from "@tanstack/react-table" +import { + flexRender, + getCoreRowModel, + getFilteredRowModel, + getSortedRowModel, + useReactTable, +} from "@tanstack/react-table" +import { + ActivityIcon, + BinaryIcon, + CheckCircleIcon, + CircleAlertIcon, + ClockIcon, + HardDriveDownloadIcon, + HardDriveIcon, + HardDriveUploadIcon, + LoaderCircleIcon, + MoreHorizontalIcon, + RefreshCwIcon, + RotateCwIcon, + XCircleIcon, + XIcon, +} from "lucide-react" +import { useCallback, useEffect, useMemo, useState } from "react" + +const ZFS_POOL_FIELDS = "id,system,name,health,size,alloc,free,scrub,details_updated,updated" + +/** Maps a zpool health string to a Badge variant. */ +function healthVariant(health: string): "success" | "warning" | "danger" | "outline" { + switch (health) { + case "ONLINE": + return "success" + case "DEGRADED": + return "warning" + case "FAULTED": + case "OFFLINE": + case "UNAVAIL": + case "REMOVED": + case "SUSPENDED": + return "danger" + default: + return "outline" + } +} + +function formatCapacity(bytes: number): string { + if (!bytes) return "-" + const { value, unit } = formatBytes(bytes) + return `${toFixedFloat(value, value >= 10 ? 1 : 2)} ${unit}` +} + +function HeaderButton({ column, name, Icon }: { column: Column; name: string; Icon: React.ElementType }) { + const isSorted = column.getIsSorted() + return ( + + ) +} + +const columns: ColumnDef[] = [ + { + accessorKey: "name", + sortingFn: (a, b) => a.original.name.localeCompare(b.original.name), + header: ({ column }) => , + cell: ({ getValue }) => {getValue() as string}, + }, + { + accessorKey: "health", + sortingFn: (a, b) => a.original.health.localeCompare(b.original.health), + header: ({ column }) => , + cell: ({ getValue }) => { + const health = (getValue() as string) || "" + return {health || t`Unknown`} + }, + }, + { + id: "size", + accessorFn: (record) => record.size, + invertSorting: true, + header: ({ column }) => , + cell: ({ getValue }) => {formatCapacity(getValue() as number)}, + }, + { + id: "used", + accessorFn: (record) => record.alloc, + invertSorting: true, + header: ({ column }) => , + cell: ({ getValue }) => {formatCapacity(getValue() as number)}, + }, + { + id: "free", + accessorFn: (record) => record.free, + invertSorting: true, + header: ({ column }) => , + cell: ({ getValue }) => {formatCapacity(getValue() as number)}, + }, + { + id: "scrub", + accessorFn: (record) => record.scrub?.state ?? "", + header: ({ column }) => , + cell: ({ row }) => { + const scrub = row.original.scrub + if (!scrub?.state) return {t`None`} + return ( + + {scrub.state} + {scrub.progress ? ` (${scrub.progress})` : ""} + + ) + }, + }, + { + id: "updated", + invertSorting: true, + accessorFn: (record) => record.details_updated || record.updated, + header: ({ column }) => , + cell: ({ getValue }) => { + const timestamp = getValue() as string + if (!timestamp) return null + const formatter = + new Date(timestamp).toDateString() === new Date().toDateString() ? hourWithSeconds : formatShortDate + return {formatter(timestamp)} + }, + }, +] + +function VdevTable({ vdevs }: { vdevs: ZfsVdev[] }) { + if (!vdevs?.length) return null + return ( +
+ + + + {t`Vdev`} + {t`State`} + {t`Read errors`} + {t`Write errors`} + {t`Checksum errors`} + + + + {vdevs.map((vdev) => ( + + {vdev.name} + + + {vdev.state ?? "-"} + + + 0 && "text-red-600 dark:text-red-400")} + > + {vdev.readErrs ?? 0} + + 0 && "text-red-600 dark:text-red-400")} + > + {vdev.writeErrs ?? 0} + + 0 && "text-red-600 dark:text-red-400" + )} + > + {vdev.checksumErrs ?? 0} + + + ))} + +
+
+ ) +} + +const datasetColumns: ColumnDef[] = [ + { + accessorKey: "name", + sortingFn: (a, b) => a.original.name.localeCompare(b.original.name), + header: ({ column }) => , + cell: ({ getValue }) => {getValue() as string}, + }, + { + id: "used", + accessorFn: (ds) => ds.used ?? 0, + invertSorting: true, + header: ({ column }) => , + cell: ({ getValue }) => {formatCapacity(getValue() as number)}, + }, + { + id: "avail", + accessorFn: (ds) => ds.avail ?? 0, + invertSorting: true, + header: ({ column }) => , + cell: ({ getValue }) => {formatCapacity(getValue() as number)}, + }, + { + accessorKey: "mount", + sortingFn: (a, b) => (a.original.mount ?? "").localeCompare(b.original.mount ?? ""), + header: ({ column }) => , + cell: ({ getValue }) => ( + {(getValue() as string) || "-"} + ), + }, +] + +function DatasetTable({ datasets }: { datasets: ZfsDataset[] }) { + const [filter, setFilter] = useState("") + const filtered = useMemo(() => { + if (!datasets) return [] + if (!filter) return datasets + const needle = filter.toLowerCase() + return datasets.filter((ds) => ds.name.toLowerCase().includes(needle)) + }, [datasets, filter]) + + const table = useReactTable({ + data: filtered, + columns: datasetColumns, + getCoreRowModel: getCoreRowModel(), + getSortedRowModel: getSortedRowModel(), + }) + + if (!datasets?.length) return null + return ( +
+
+

+ Datasets +

+
+ setFilter(e.target.value)} + className="px-4 w-full" + /> + {filter && ( + + )} +
+
+
+ + + {table.getHeaderGroups().map((headerGroup) => ( + + {headerGroup.headers.map((header) => ( + + {header.isPlaceholder ? null : flexRender(header.column.columnDef.header, header.getContext())} + + ))} + + ))} + + + {table.getRowModel().rows.map((row) => ( + + {row.getVisibleCells().map((cell) => ( + + {flexRender(cell.column.columnDef.cell, cell.getContext())} + + ))} + + ))} + +
+
+
+ ) +} + +function PoolSheet({ + poolId, + open, + onOpenChange, +}: { + poolId: string | null + open: boolean + onOpenChange: (open: boolean) => void +}) { + const [pool, setPool] = useState(null) + const [isLoading, setIsLoading] = useState(false) + + useEffect(() => { + let active = true + if (!poolId) { + setPool(null) + return + } + // Only fetch when opening, not when closing (keeps data visible during close animation) + if (!open) return + setIsLoading(true) + pb.collection("zfs_pools") + .getOne(poolId) + .then((record) => active && setPool(record as ZfsPoolRecord)) + .catch(() => active && setPool(null)) + .finally(() => active && setIsLoading(false)) + return () => { + active = false + } + }, [open, poolId]) + + const health = pool?.health || "" + const healthVariantValue = healthVariant(health) + const HealthIcon = + healthVariantValue === "success" + ? CheckCircleIcon + : healthVariantValue === "warning" + ? CircleAlertIcon + : XCircleIcon + + return ( + + + + + {pool ? pool.name : t`ZFS Pool`} + {pool && {health}} + + + {pool?.size ? formatCapacity(pool.size) : null} + {pool?.alloc ? ( + <> + + + Used: {formatCapacity(pool.alloc)} + + + ) : null} + {pool?.free ? ( + <> + + + Free: {formatCapacity(pool.free)} + + + ) : null} + + +
+ {isLoading ? ( +
+ +
+ ) : ( + <> + {pool && health && ( + + + + Pool Health: {health} + + {pool.scrub?.state && ( + + Scrub: {pool.scrub.state} + {pool.scrub.progress ? ` (${pool.scrub.progress})` : ""} + {pool.scrub.errors ? `, ${pool.scrub.errors} errors` : ""} + + )} + + )} + {pool?.vdevs?.length || pool?.datasets?.length ? ( + <> + {pool.vdevs?.length ? : null} + + + ) : ( + !isLoading && ( +
+ No detail data for this pool. +
+ ) + )} + + )} +
+ + + ) +} + +export default function ZfsTable({ systemId }: { systemId?: string }) { + const [zfsPools, setZfsPools] = useState() + const [globalFilter, setGlobalFilter] = useState("") + const [activePoolId, setActivePoolId] = useState(null) + const [sheetOpen, setSheetOpen] = useState(false) + const [refreshingId, setRefreshingId] = useState(null) + + useEffect(() => { + let disposed = false + let unsubscribe: () => void = () => {} + // fetch initial records + pb.collection("zfs_pools") + .getFullList({ + filter: systemId ? pb.filter("system={:id}", { id: systemId }) : "", + sort: "name", + fields: ZFS_POOL_FIELDS, + }) + .then((records) => !disposed && setZfsPools(records)) + .catch((error) => console.error("Failed to fetch ZFS pools:", error)) + + // subscribe to realtime updates + const pbOptions = systemId ? { filter: `system="${systemId}"` } : undefined + ;(async () => { + try { + const unsubscribeNow = await pb.collection("zfs_pools").subscribe( + "*", + (event) => { + const record = event.record as ZfsPoolRecord + setZfsPools((current) => { + const pools = current ?? [] + const matchesSystemScope = !systemId || record.system === systemId + if (event.action === "delete") { + return pools.filter((pool) => pool.id !== record.id) + } + if (!matchesSystemScope) { + return pools.filter((pool) => pool.id !== record.id) + } + const existingIndex = pools.findIndex((pool) => pool.id === record.id) + if (existingIndex === -1) { + return [record, ...pools] + } + const next = [...pools] + next[existingIndex] = record + return next + }) + }, + pbOptions + ) + if (disposed) { + unsubscribeNow() + } else { + unsubscribe = unsubscribeNow + } + } catch (error) { + console.error("Failed to subscribe to ZFS pool updates:", error) + } + })() + + return () => { + disposed = true + unsubscribe?.() + } + }, [systemId]) + + const refreshSystem = useCallback(async (systemId: string) => { + try { + await pb.send("/api/beszel/zfs/refresh", { + method: "POST", + query: { system: systemId }, + }) + } catch (error) { + console.error("Failed to refresh ZFS pools:", error) + } + }, []) + + const handleRowRefresh = useCallback( + async (pool: ZfsPoolRecord) => { + if (!pool.system) return + setRefreshingId(pool.id) + try { + await refreshSystem(pool.system) + } finally { + setRefreshingId((id) => (id === pool.id ? null : id)) + } + }, + [refreshSystem] + ) + + const actionColumn = useMemo>( + () => ({ + id: "actions", + enableSorting: false, + header: () => ( + + Actions + + ), + cell: ({ row }) => { + const pool = row.original + const isRowRefreshing = refreshingId === pool.id + + return ( +
+ + + + + event.stopPropagation()}> + { + event.stopPropagation() + handleRowRefresh(pool) + }} + disabled={isRowRefreshing} + > + + Refresh + + + +
+ ) + }, + }), + [refreshingId, handleRowRefresh] + ) + + const tableColumns = useMemo(() => { + return isReadOnlyUser() ? columns : [...columns, actionColumn] + }, [actionColumn]) + + const table = useReactTable({ + data: zfsPools || ([] as ZfsPoolRecord[]), + columns: tableColumns, + getCoreRowModel: getCoreRowModel(), + getSortedRowModel: getSortedRowModel(), + getFilteredRowModel: getFilteredRowModel(), + state: { globalFilter }, + onGlobalFilterChange: setGlobalFilter, + globalFilterFn: (row, _columnId, filterValue) => { + const pool = row.original + const searchString = `${pool.name} ${pool.health ?? ""}`.toLowerCase() + return (filterValue as string) + .toLowerCase() + .split(" ") + .every((term) => searchString.includes(term)) + }, + }) + const rows = table.getRowModel().rows + + // Hide the table on system pages if there's no data + if (systemId && !zfsPools?.length && !globalFilter) { + return null + } + + const openSheet = (pool: ZfsPoolRecord) => { + setActivePoolId(pool.id) + setSheetOpen(true) + } + + return ( +
+ + +
+
+ ZFS + + Click on a pool to view vdev and dataset details. + +
+
+ setGlobalFilter(event.target.value)} + className="px-4 w-full max-w-full md:w-64" + /> + {globalFilter && ( + + )} +
+
+ +
+ + + {table.getHeaderGroups().map((headerGroup) => ( + + {headerGroup.headers.map((header) => ( + + {header.isPlaceholder ? null : flexRender(header.column.columnDef.header, header.getContext())} + + ))} + + ))} + + + {rows.map((row) => ( + openSheet(row.original)} + > + {row.getVisibleCells().map((cell) => ( + {flexRender(cell.column.columnDef.cell, cell.getContext())} + ))} + + ))} + +
+
+ + +
+ ) +} diff --git a/internal/site/src/types.d.ts b/internal/site/src/types.d.ts index a35cb5a1..3ef5293c 100644 --- a/internal/site/src/types.d.ts +++ b/internal/site/src/types.d.ts @@ -151,6 +151,8 @@ export interface SystemStats { f?: Record /** extra filesystems */ efs?: Record + /** ZFS pool metrics */ + z?: Record /** GPU data */ g?: Record /** battery percent and state */ @@ -178,6 +180,56 @@ export interface GPUData { e?: Record } +export interface ZfsPool { + /** total capacity (GiB) */ + d: number + /** allocated (GiB) */ + du: number + /** read throughput (bytes/s) */ + rb?: number + /** write throughput (bytes/s) */ + wb?: number + /** health: ONLINE, DEGRADED, FAULTED, ... */ + h?: string +} + +export interface ZfsScrub { + /** NONE, SCANNING, FINISHED, CANCELED */ + state?: string + /** progress while scanning, e.g. "10.00%" */ + progress?: string + errors?: number +} + +export interface ZfsVdev { + name: string + state?: string + readErrs?: number + writeErrs?: number + checksumErrs?: number +} + +export interface ZfsDataset { + name: string + used?: number + avail?: number + mount?: string +} + +export interface ZfsPoolRecord extends RecordModel { + system: string + name: string + health: string + size: number + alloc: number + free: number + scrub: ZfsScrub | null + vdevs: ZfsVdev[] | null + datasets: ZfsDataset[] | null + details_updated: string + updated: string +} + export interface ExtraFsStats { /** disk size (gb) */ d: number diff --git a/readme.md b/readme.md index 8685a8a3..57e4dea2 100644 --- a/readme.md +++ b/readme.md @@ -16,6 +16,7 @@ It has a friendly web interface, simple configuration, and is ready to use out o - **Lightweight**: Smaller and less resource-intensive than leading solutions. - **Simple**: Easy setup with little manual configuration required. - **Docker stats**: Tracks CPU, memory, and network usage history for each container. +- **ZFS**: Tracks pool capacity, health, and I/O, plus per-dataset usage. - **Alerts**: Configurable alerts for CPU, memory, disk, bandwidth, temperature, fan speed, load average, and status. - **Multi-user**: Users manage their own systems. Admins can share systems across users. - **OAuth / OIDC**: Supports many OAuth2 providers. Password auth can be disabled. @@ -53,6 +54,7 @@ The [quick start guide](https://beszel.dev/guide/getting-started) and other docu - **Battery** - Host system battery charge. - **Containers** - Status and metrics of all running Docker / Podman containers. - **S.M.A.R.T.** - Host system disk health (includes eMMC wear/EOL and Linux mdraid array health via sysfs when available). +- **ZFS** - Pool capacity, usage, health, I/O throughput, scrub status, and per-dataset usage. ## Help and discussion