beszel-ipv6/agent/zfs_pool.go

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
}