fix(agent): measure disk usage within the server data volume

agent/cmd/main.go

@@ -244,6 +244,13 @@ func main() {
 
 	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer cancel()
+	usageReader := usage.New(usage.Config{
+		DataDir:            dataRoot,
+		ProcMode:           envOr("GAMEPLANE_USAGE_PROC", "") == "1",
+		CPULimitMillicores: envInt("GAMEPLANE_CPU_LIMIT_MILLICORES"),
+		MemLimitBytes:      envInt("GAMEPLANE_MEM_LIMIT_BYTES"),
+	})
+	go usageReader.MonitorDisk(ctx)
 
 	// Prometheus metrics for per-server resource usage and player counts.
 	// Registered with the default registry so promhttp.Handler() picks it up.
@@ -268,12 +275,7 @@ func main() {
 			}
 			return nil
 		}(),
-		Usage: usage.New(usage.Config{
-			DataDir:            dataRoot,
-			ProcMode:           envOr("GAMEPLANE_USAGE_PROC", "") == "1",
-			CPULimitMillicores: envInt("GAMEPLANE_CPU_LIMIT_MILLICORES"),
-			MemLimitBytes:      envInt("GAMEPLANE_MEM_LIMIT_BYTES"),
-		}),
+		Usage:   usageReader,
 		Metrics: m,
 	})
 

agent/internal/heartbeat/heartbeat.go

@@ -187,7 +187,7 @@ func sendOnce(ctx context.Context, dyn dynamic.Interface, cfg Config) error {
 		agent["memoryBytes"] = nullable(s.MemoryBytes, s.MemoryKnown)
 		agent["memoryLimitBytes"] = nullable(s.MemoryLimitBytes, s.MemoryLimitKnown)
 		agent["diskUsedBytes"] = nullable(s.DiskUsedBytes, s.DiskKnown)
-		agent["diskTotalBytes"] = nullable(s.DiskTotalBytes, s.DiskKnown)
+		agent["diskTotalBytes"] = nullable(s.DiskTotalBytes, s.DiskTotalKnown)
 		// Update metrics with usage data.
 		if cfg.Metrics != nil {
 			metricsSnap.CPUMillicores = s.CPUMillicores
@@ -199,8 +199,9 @@ func sendOnce(ctx context.Context, dyn dynamic.Interface, cfg Config) error {
 			metricsSnap.MemoryLimitBytes = s.MemoryLimitBytes
 			metricsSnap.MemoryLimitKnown = s.MemoryLimitKnown
 			metricsSnap.DiskUsedBytes = s.DiskUsedBytes
-			metricsSnap.DiskTotalBytes = s.DiskTotalBytes
 			metricsSnap.DiskKnown = s.DiskKnown
+			metricsSnap.DiskTotalKnown = s.DiskTotalKnown
+			metricsSnap.DiskTotalBytes = s.DiskTotalBytes
 		}
 	}
 

agent/internal/heartbeat/heartbeat_test.go

@@ -155,7 +155,7 @@ func TestSendOnce_EmitsKnownUsage(t *testing.T) {
 			CPULimitMillicores: 2000, CPULimitKnown: true,
 			MemoryBytes: 1024, MemoryKnown: true,
 			MemoryLimitBytes: 4096, MemoryLimitKnown: true,
-			DiskUsedBytes: 10, DiskTotalBytes: 100, DiskKnown: true,
+			DiskUsedBytes: 10, DiskKnown: true, DiskTotalBytes: 100, DiskTotalKnown: true,
 		}},
 	})
 	if err != nil {
@@ -452,8 +452,8 @@ func TestSendOnce_PopulatesMetricsSnapshot_PlayersAndUsage(t *testing.T) {
 			MemoryLimitBytes:   4096,
 			MemoryLimitKnown:   true,
 			DiskUsedBytes:      100,
-			DiskTotalBytes:     1000,
-			DiskKnown:          true,
+			DiskTotalBytes:     1000, DiskTotalKnown: true,
+			DiskKnown: true,
 		}},
 		Metrics: metricsObj,
 	}

agent/internal/metrics/metrics.go

@@ -28,6 +28,7 @@ type Snapshot struct {
 	// Disk bytes: used and total. Only emitted when known=true.
 	DiskUsedBytes  int64
 	DiskTotalBytes int64
+	DiskTotalKnown bool
 	DiskKnown      bool
 
 	// Player counts. PlayersOnline is emitted when known=true.
@@ -111,6 +112,8 @@ func (m *Metrics) Collect(ch chan<- prometheus.Metric) {
 	// Disk: emit only when known.
 	if snap.DiskKnown {
 		ch <- prometheus.MustNewConstMetric(diskUsedDesc, prometheus.GaugeValue, float64(snap.DiskUsedBytes), labels...)
+	}
+	if snap.DiskTotalKnown {
 		ch <- prometheus.MustNewConstMetric(diskTotalDesc, prometheus.GaugeValue, float64(snap.DiskTotalBytes), labels...)
 	}
 
@@ -155,7 +158,7 @@ var (
 	)
 	diskTotalDesc = prometheus.NewDesc(
 		"gameplane_agent_disk_total_bytes",
-		"Total disk capacity of the game's data directory, in bytes. Emitted only when readable.",
+		"Observed capacity scoped to the game's data directory, in bytes. Emitted only when available.",
 		[]string{"server", "namespace", "template", "game"}, nil,
 	)
 	playersOnlineDesc = prometheus.NewDesc(

agent/internal/metrics/metrics_test.go

@@ -24,6 +24,7 @@ func TestMetricsEmitsWhenKnown(t *testing.T) {
 		MemoryLimitKnown:   true,
 		DiskUsedBytes:      1024 * 1024 * 100,  // 100 MiB
 		DiskTotalBytes:     1024 * 1024 * 1024, // 1 GiB
+		DiskTotalKnown:     true,
 		DiskKnown:          true,
 		PlayersOnline:      5,
 		PlayersOnlineKnown: true,
@@ -257,6 +258,7 @@ func TestMetricsZeroValuesEmitted(t *testing.T) {
 		MemoryBytes:        0,
 		MemoryKnown:        true,
 		DiskUsedBytes:      0,
+		DiskTotalKnown:     true,
 		DiskKnown:          true,
 		PlayersOnline:      0,
 		PlayersOnlineKnown: true,
@@ -438,6 +440,7 @@ func TestMetricsDiskZeroTotal(t *testing.T) {
 	snap := Snapshot{
 		DiskUsedBytes:  0,
 		DiskTotalBytes: 0, // Edge case: zero total (e.g., mount failed)
+		DiskTotalKnown: true,
 		DiskKnown:      true,
 		ServerName:     "test-server",
 		Namespace:      "games",
@@ -592,6 +595,7 @@ func TestMetricsAllMetricsIndependent(t *testing.T) {
 		// Disk: known
 		DiskUsedBytes:  500 * 1024 * 1024,
 		DiskTotalBytes: 1024 * 1024 * 1024,
+		DiskTotalKnown: true,
 		DiskKnown:      true,
 		// Players: online known, max unknown
 		PlayersOnline:      8,

agent/internal/usage/disk.go

@@ -0,0 +1,205 @@
+package usage
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"os"
+	"strconv"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+const (
+	diskInterval   = 5 * time.Minute
+	diskTimeout    = time.Minute
+	maxDiskEntries = 2_000_000
+	maxDiskDepth   = 128
+)
+
+// MonitorDisk runs one scan at a time until ctx is canceled. A shared
+// filesystem has no per-directory capacity. The reader publishes allocated
+// bytes only after the entire scan completes, and clears a failed sample.
+func (r *Reader) MonitorDisk(ctx context.Context) { r.monitorDisk(ctx, volumeUsage) }
+
+func (r *Reader) monitorDisk(ctx context.Context, scan func(context.Context, string, int) (diskSample, error)) {
+	if r.cfg.DataDir == "" {
+		return
+	}
+	for ctx.Err() == nil {
+		scanCtx, cancel := context.WithTimeout(ctx, diskTimeout)
+		sample, err := scan(scanCtx, r.cfg.DataDir, maxDiskEntries)
+		// A scan and its timer can become runnable together. Observe the
+		// deadline itself even if cancellation has not been delivered yet.
+		deadline, _ := scanCtx.Deadline()
+		completed := err == nil && scanCtx.Err() == nil && time.Now().Before(deadline)
+		cancel()
+		r.mu.Lock()
+		r.diskMeasuredAt = time.Time{}
+		if completed {
+			r.diskBytes = sample.used
+			r.diskCapacity = sample.capacity
+			r.diskMeasuredAt = r.cfg.now()
+		}
+		r.mu.Unlock()
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(diskInterval):
+		}
+	}
+}
+
+func (r *Reader) readDisk(s *Sample) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if !r.diskMeasuredAt.IsZero() && r.cfg.now().Sub(r.diskMeasuredAt) < diskInterval+diskTimeout {
+		s.DiskUsedBytes = r.diskBytes
+		s.DiskKnown = true
+		s.DiskTotalBytes = r.diskCapacity
+		s.DiskTotalKnown = r.diskCapacity > 0
+	}
+}
+
+type diskSample struct{ used, capacity int64 }
+
+func volumeUsage(ctx context.Context, root string, limit int) (diskSample, error) {
+	fd, err := unix.Open(root, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0)
+	if err != nil {
+		return diskSample{}, err
+	}
+	dir := os.NewFile(uintptr(fd), root)
+	defer func() { _ = dir.Close() }()
+	var fs unix.Statfs_t
+	if err := unix.Fstatfs(fd, &fs); err == nil && fs.Type == unix.CEPH_SUPER_MAGIC {
+		// Kernel CephFS exposes recursive usage and quotas as virtual xattrs on
+		// this directory. An inherited/absent quota cannot establish capacity.
+		if s, ok := cephDiskUsage(func(key string) (int64, error) { return diskAttribute(fd, key) }); ok {
+			return s, nil
+		}
+	}
+	used, err := allocatedDiskUsage(ctx, dir, limit)
+	return diskSample{used: used}, err
+}
+
+// CephFS exposes the counters as decimal integers on the opened directory.
+func diskAttribute(fd int, key string) (int64, error) {
+	var value [32]byte
+	n, err := unix.Fgetxattr(fd, key, value[:])
+	if err != nil {
+		return 0, err
+	}
+	return strconv.ParseInt(string(value[:n]), 10, 64)
+}
+
+func cephDiskUsage(read func(string) (int64, error)) (diskSample, bool) {
+	used, err := read("ceph.dir.rbytes")
+	if err != nil || used < 0 {
+		return diskSample{}, false
+	}
+	quota, err := read("ceph.quota.max_bytes")
+	if err != nil || quota < 0 {
+		quota = 0
+	}
+	return diskSample{used: used, capacity: quota}, true
+}
+
+type diskWalk struct {
+	ctx       context.Context
+	remaining int
+	// Single-link files need no entry. NO_XDEV confines inode numbers to one mount.
+	hardlinks map[uint64]struct{}
+	bytes     int64
+}
+
+func allocatedDiskUsage(ctx context.Context, dir *os.File, limit int) (int64, error) {
+	fd := int(dir.Fd())
+	walk := diskWalk{ctx: ctx, remaining: limit, hardlinks: make(map[uint64]struct{})}
+	if err := walk.count(fd); err != nil {
+		return 0, err
+	}
+	if err := walk.dir(dir, 0); err != nil {
+		return 0, err
+	}
+	return walk.bytes, nil
+}
+
+func (w *diskWalk) count(fd int) error {
+	var st unix.Stat_t
+	if err := unix.Fstat(fd, &st); err != nil {
+		return err
+	}
+	if st.Mode&unix.S_IFMT != unix.S_IFDIR && st.Nlink > 1 {
+		if _, ok := w.hardlinks[st.Ino]; ok {
+			return nil
+		}
+		w.hardlinks[st.Ino] = struct{}{}
+	}
+	if st.Blocks < 0 || st.Blocks > (math.MaxInt64-w.bytes)/512 {
+		return errors.New("allocated byte count overflow")
+	}
+	w.bytes += st.Blocks * 512
+	return nil
+}
+
+func (w *diskWalk) dir(dir *os.File, depth int) error {
+	if depth > maxDiskDepth {
+		return errors.New("disk scan depth exceeded")
+	}
+	for {
+		if err := w.ctx.Err(); err != nil {
+			return err
+		}
+		names, err := dir.Readdirnames(128)
+		if err != nil && !errors.Is(err, io.EOF) {
+			return err
+		}
+		for _, name := range names {
+			if err := w.ctx.Err(); err != nil {
+				return err
+			}
+			w.remaining--
+			if w.remaining < 0 {
+				return errors.New("disk scan entry limit exceeded")
+			}
+			if err := w.entry(int(dir.Fd()), name, depth); err != nil {
+				return err
+			}
+		}
+		if errors.Is(err, io.EOF) {
+			return nil
+		}
+	}
+}
+
+func (w *diskWalk) entry(parent int, name string, depth int) error {
+	// O_PATH avoids opening FIFOs or devices for I/O. NO_XDEV also excludes
+	// bind mounts on the same device; NOFOLLOW counts the link itself.
+	fd, err := unix.Openat2(parent, name, &unix.OpenHow{Flags: unix.O_PATH | unix.O_CLOEXEC | unix.O_NOFOLLOW, Resolve: unix.RESOLVE_BENEATH | unix.RESOLVE_NO_XDEV | unix.RESOLVE_NO_MAGICLINKS})
+	// Entries that disappeared after readdir no longer occupy this path.
+	if errors.Is(err, unix.EXDEV) || errors.Is(err, unix.ENOENT) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("open disk entry: %w", err)
+	}
+	defer func() { _ = unix.Close(fd) }()
+	if err := w.count(fd); err != nil {
+		return err
+	}
+	// Reopen the pinned inode as a readable directory. ENOTDIR identifies
+	// leaves without trusting the directory entry type or reopening its name.
+	child, err := unix.Openat(fd, ".", unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC, 0)
+	if errors.Is(err, unix.ENOTDIR) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	dir := os.NewFile(uintptr(child), name)
+	defer func() { _ = dir.Close() }()
+	return w.dir(dir, depth+1)
+}