feat: add ZFS monitoring (#2209)

- track pool capacity, health, I/O, scrub status, and vdev errors
- report dataset usage and correct ZFS filesystem metrics
- add pool charts, detail views, refresh controls, and health alerts
- persist pool details and include ZFS usage in disk alerts
- support configurable detail intervals and legacy agent compatibility

---------

Co-authored-by: hank <hank@henrygd.me>

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
 }
 

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("")

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
+}

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"))
+}

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 }

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.