fix(hostmetrics): dedupe ZFS pools for usable storage

Related to #718

internal/hostmetrics/collector.go

@@ -97,6 +97,7 @@ func collectDisks(ctx context.Context) []agentshost.Disk {
 
 	disks := make([]agentshost.Disk, 0, len(partitions))
 	seen := make(map[string]struct{}, len(partitions))
+	zfsDatasets := make([]zfsDatasetUsage, 0)
 
 	for _, part := range partitions {
 		if part.Mountpoint == "" {
@@ -115,6 +116,26 @@ func collectDisks(ctx context.Context) []agentshost.Disk {
 			continue
 		}
 
+		if part.Fstype == "zfs" {
+			pool := zfsPoolFromDevice(part.Device)
+			if pool == "" {
+				continue
+			}
+			if fsfilters.ShouldIgnoreReadOnlyFilesystem(part.Fstype, usage.Total, usage.Used) {
+				continue
+			}
+
+			zfsDatasets = append(zfsDatasets, zfsDatasetUsage{
+				Pool:       pool,
+				Dataset:    part.Device,
+				Mountpoint: part.Mountpoint,
+				Total:      usage.Total,
+				Used:       usage.Used,
+				Free:       usage.Free,
+			})
+			continue
+		}
+
 		// Skip read-only filesystems like squashfs (snap mounts), erofs, iso9660, etc.
 		// These are immutable and always report near-full usage, which causes false alerts.
 		// See issues #505 (Home Assistant OS) and #690 (Ubuntu snap mounts).
@@ -134,6 +155,8 @@ func collectDisks(ctx context.Context) []agentshost.Disk {
 		})
 	}
 
+	disks = append(disks, summarizeZFSPools(ctx, zfsDatasets)...)
+
 	sort.Slice(disks, func(i, j int) bool { return disks[i].Mountpoint < disks[j].Mountpoint })
 	return disks
 }

internal/hostmetrics/zfs.go

@@ -0,0 +1,297 @@
+package hostmetrics
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"os/exec"
+	"sort"
+	"strconv"
+	"strings"
+
+	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
+)
+
+// zpoolStats represents capacity data reported by `zpool list`.
+type zpoolStats struct {
+	Size  uint64
+	Alloc uint64
+	Free  uint64
+}
+
+// zfsDatasetUsage preserves per-dataset usage so we can reconcile pools later.
+type zfsDatasetUsage struct {
+	Pool       string
+	Dataset    string
+	Mountpoint string
+	Total      uint64
+	Used       uint64
+	Free       uint64
+}
+
+var queryZpoolStats = fetchZpoolStats
+
+func summarizeZFSPools(ctx context.Context, datasets []zfsDatasetUsage) []agentshost.Disk {
+	if len(datasets) == 0 {
+		return nil
+	}
+
+	pools := uniqueZFSPools(datasets)
+	if len(pools) == 0 {
+		return nil
+	}
+
+	bestDatasets := bestZFSPoolDatasets(datasets)
+	mountpoints := bestZFSMountpoints(datasets)
+	stats, err := queryZpoolStats(ctx, pools)
+	if err == nil && len(stats) > 0 {
+		return disksFromZpoolStats(pools, stats, mountpoints, bestDatasets)
+	}
+
+	return fallbackZFSDisks(bestDatasets, mountpoints)
+}
+
+func disksFromZpoolStats(
+	pools []string,
+	stats map[string]zpoolStats,
+	mountpoints map[string]string,
+	bestDatasets map[string]zfsDatasetUsage,
+) []agentshost.Disk {
+	disks := make([]agentshost.Disk, 0, len(pools))
+
+	for _, pool := range pools {
+		stat, ok := stats[pool]
+		mp := mountpoints[pool]
+		if mp == "" {
+			mp = fmt.Sprintf("zpool:%s", pool)
+		}
+
+		if ok && stat.Size > 0 {
+			usage := clampPercent(calculatePercent(stat.Size, stat.Alloc))
+			disks = append(disks, agentshost.Disk{
+				Device:     pool,
+				Mountpoint: mp,
+				Filesystem: "zfs",
+				Type:       "zfs",
+				TotalBytes: int64(stat.Size),
+				UsedBytes:  int64(stat.Alloc),
+				FreeBytes:  int64(stat.Free),
+				Usage:      usage,
+			})
+			continue
+		}
+
+		if ds, ok := bestDatasets[pool]; ok && ds.Total > 0 {
+			usage := clampPercent(calculatePercent(ds.Total, ds.Used))
+			disks = append(disks, agentshost.Disk{
+				Device:     pool,
+				Mountpoint: mp,
+				Filesystem: "zfs",
+				Type:       "zfs",
+				TotalBytes: int64(ds.Total),
+				UsedBytes:  int64(ds.Used),
+				FreeBytes:  int64(ds.Free),
+				Usage:      usage,
+			})
+		}
+	}
+
+	return disks
+}
+
+func fallbackZFSDisks(bestDatasets map[string]zfsDatasetUsage, mountpoints map[string]string) []agentshost.Disk {
+	if len(bestDatasets) == 0 {
+		return nil
+	}
+
+	pools := make([]string, 0, len(bestDatasets))
+	for pool := range bestDatasets {
+		pools = append(pools, pool)
+	}
+	sort.Strings(pools)
+
+	disks := make([]agentshost.Disk, 0, len(pools))
+	for _, pool := range pools {
+		ds := bestDatasets[pool]
+		if ds.Total == 0 {
+			continue
+		}
+
+		mp := mountpoints[pool]
+		if mp == "" {
+			mp = fmt.Sprintf("zpool:%s", pool)
+		}
+
+		usage := clampPercent(calculatePercent(ds.Total, ds.Used))
+		disks = append(disks, agentshost.Disk{
+			Device:     pool,
+			Mountpoint: mp,
+			Filesystem: "zfs",
+			Type:       "zfs",
+			TotalBytes: int64(ds.Total),
+			UsedBytes:  int64(ds.Used),
+			FreeBytes:  int64(ds.Free),
+			Usage:      usage,
+		})
+	}
+
+	return disks
+}
+
+func fetchZpoolStats(ctx context.Context, pools []string) (map[string]zpoolStats, error) {
+	if len(pools) == 0 {
+		return nil, nil
+	}
+
+	path, err := exec.LookPath("zpool")
+	if err != nil {
+		return nil, err
+	}
+
+	args := []string{"list", "-Hp", "-o", "name,size,allocated,free"}
+	args = append(args, pools...)
+
+	cmd := exec.CommandContext(ctx, path, args...)
+	output, err := cmd.Output()
+	if err != nil {
+		return nil, err
+	}
+
+	return parseZpoolList(output)
+}
+
+func parseZpoolList(output []byte) (map[string]zpoolStats, error) {
+	stats := make(map[string]zpoolStats)
+	scanner := bufio.NewScanner(bytes.NewReader(output))
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == "" {
+			continue
+		}
+
+		fields := strings.Split(line, "\t")
+		if len(fields) < 4 {
+			continue
+		}
+
+		size, err := strconv.ParseUint(fields[1], 10, 64)
+		if err != nil {
+			continue
+		}
+		alloc, err := strconv.ParseUint(fields[2], 10, 64)
+		if err != nil {
+			continue
+		}
+		free, err := strconv.ParseUint(fields[3], 10, 64)
+		if err != nil {
+			continue
+		}
+
+		stats[fields[0]] = zpoolStats{
+			Size:  size,
+			Alloc: alloc,
+			Free:  free,
+		}
+	}
+
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
+	if len(stats) == 0 {
+		return nil, fmt.Errorf("zpool list returned no usable data")
+	}
+	return stats, nil
+}
+
+func uniqueZFSPools(datasets []zfsDatasetUsage) []string {
+	set := make(map[string]struct{}, len(datasets))
+	for _, ds := range datasets {
+		if ds.Pool != "" {
+			set[ds.Pool] = struct{}{}
+		}
+	}
+	if len(set) == 0 {
+		return nil
+	}
+
+	pools := make([]string, 0, len(set))
+	for pool := range set {
+		pools = append(pools, pool)
+	}
+	sort.Strings(pools)
+	return pools
+}
+
+func bestZFSMountpoints(datasets []zfsDatasetUsage) map[string]string {
+	mounts := make(map[string]string, len(datasets))
+	scores := make(map[string]int, len(datasets))
+
+	for _, ds := range datasets {
+		if ds.Pool == "" || ds.Mountpoint == "" {
+			continue
+		}
+
+		score := zfsMountpointScore(ds)
+		if current, ok := scores[ds.Pool]; ok && score >= current {
+			continue
+		}
+		scores[ds.Pool] = score
+		mounts[ds.Pool] = ds.Mountpoint
+	}
+
+	return mounts
+}
+
+func zfsMountpointScore(ds zfsDatasetUsage) int {
+	if ds.Dataset != "" && !strings.Contains(ds.Dataset, "/") {
+		return 0
+	}
+	path := strings.Trim(ds.Mountpoint, "/")
+	if path == "" {
+		return 1
+	}
+	return 1 + strings.Count(path, "/")
+}
+
+func zfsPoolFromDevice(device string) string {
+	device = strings.TrimSpace(device)
+	if device == "" {
+		return ""
+	}
+	if idx := strings.Index(device, "/"); idx >= 0 {
+		return device[:idx]
+	}
+	return device
+}
+
+func calculatePercent(total, used uint64) float64 {
+	if total == 0 {
+		return 0
+	}
+	return (float64(used) / float64(total)) * 100
+}
+
+func clampPercent(value float64) float64 {
+	switch {
+	case value < 0:
+		return 0
+	case value > 100:
+		return 100
+	default:
+		return value
+	}
+}
+
+func bestZFSPoolDatasets(datasets []zfsDatasetUsage) map[string]zfsDatasetUsage {
+	best := make(map[string]zfsDatasetUsage)
+	for _, ds := range datasets {
+		if ds.Pool == "" {
+			continue
+		}
+		if current, ok := best[ds.Pool]; !ok || ds.Total > current.Total {
+			best[ds.Pool] = ds
+		}
+	}
+	return best
+}

internal/hostmetrics/zfs_test.go

@@ -0,0 +1,90 @@
+package hostmetrics
+
+import (
+	"context"
+	"testing"
+)
+
+func TestSummarizeZFSPoolsUsesZpoolStats(t *testing.T) {
+	originalQuery := queryZpoolStats
+	t.Cleanup(func() { queryZpoolStats = originalQuery })
+
+	queryZpoolStats = func(ctx context.Context, pools []string) (map[string]zpoolStats, error) {
+		return map[string]zpoolStats{
+			"bpool": {Size: 1000, Alloc: 400, Free: 600},
+			"tank":  {Size: 5000, Alloc: 2000, Free: 3000},
+		}, nil
+	}
+
+	datasets := []zfsDatasetUsage{
+		{Pool: "tank", Dataset: "tank", Mountpoint: "/mnt/tank", Total: 5000, Used: 2000, Free: 3000},
+		{Pool: "tank", Dataset: "tank/home", Mountpoint: "/mnt/tank/home", Total: 4500, Used: 1500, Free: 3000},
+		{Pool: "bpool", Dataset: "bpool/ROOT/debian", Mountpoint: "/boot", Total: 800, Used: 200, Free: 600},
+	}
+
+	disks := summarizeZFSPools(context.Background(), datasets)
+	if len(disks) != 2 {
+		t.Fatalf("expected 2 disks, got %d", len(disks))
+	}
+
+	if disks[0].Device != "bpool" || disks[1].Device != "tank" {
+		t.Fatalf("unexpected disk order: %+v", disks)
+	}
+
+	tank := disks[1]
+	if tank.Mountpoint != "/mnt/tank" {
+		t.Errorf("expected tank mountpoint /mnt/tank, got %s", tank.Mountpoint)
+	}
+	if tank.TotalBytes != 5000 || tank.UsedBytes != 2000 || tank.FreeBytes != 3000 {
+		t.Errorf("unexpected tank capacity %+v", tank)
+	}
+	if tank.Usage < 39.9 || tank.Usage > 40.1 {
+		t.Errorf("expected tank usage ~40%%, got %.2f", tank.Usage)
+	}
+}
+
+func TestSummarizeZFSPoolsFallback(t *testing.T) {
+	originalQuery := queryZpoolStats
+	t.Cleanup(func() { queryZpoolStats = originalQuery })
+
+	queryZpoolStats = func(ctx context.Context, pools []string) (map[string]zpoolStats, error) {
+		return nil, context.DeadlineExceeded
+	}
+
+	datasets := []zfsDatasetUsage{
+		{Pool: "tank", Dataset: "tank", Mountpoint: "/mnt/tank", Total: 5000, Used: 2000, Free: 3000},
+		{Pool: "tank", Dataset: "tank/home", Mountpoint: "/mnt/tank/home", Total: 4500, Used: 1500, Free: 3000},
+		{Pool: "bpool", Dataset: "bpool/ROOT/debian", Mountpoint: "/boot", Total: 800, Used: 200, Free: 600},
+	}
+
+	disks := summarizeZFSPools(context.Background(), datasets)
+	if len(disks) != 2 {
+		t.Fatalf("expected 2 disks, got %d", len(disks))
+	}
+
+	tank := disks[1]
+	if tank.TotalBytes != 5000 || tank.UsedBytes != 2000 || tank.FreeBytes != 3000 {
+		t.Errorf("tank fallback totals incorrect: %+v", tank)
+	}
+	if tank.Usage < 39.9 || tank.Usage > 40.1 {
+		t.Errorf("expected tank usage ~40%%, got %.2f", tank.Usage)
+	}
+}
+
+func TestZFSPoolFromDevice(t *testing.T) {
+	tests := []struct {
+		device string
+		want   string
+	}{
+		{device: "tank/ROOT/default", want: "tank"},
+		{device: "vault", want: "vault"},
+		{device: "", want: ""},
+	}
+
+	for _, tt := range tests {
+		got := zfsPoolFromDevice(tt.device)
+		if got != tt.want {
+			t.Errorf("zfsPoolFromDevice(%q) = %q, want %q", tt.device, got, tt.want)
+		}
+	}
+}