feat(resource): report node utilization on the resource endpoint

README.md

@@ -65,6 +65,8 @@ The cgroup provider is read-only and works with both cgroup v1 and v2. It does
 not enable controllers or create child cgroups, including when experimental
 cgroup-disabled execution is selected.
 
+The same `/resource` response includes cached CPU, memory, PID, FD, and filestore disk utilization for advisory scheduling protection. See [resource utilization](doc/resource-utilization.md) for the wire format and sampling scope.
+
 See [test/e2e/README.md](test/e2e/README.md#gpu-debug-image) for the validated
 GPU debug image and manual CUDA test.
 

doc/resource-utilization.md

@@ -0,0 +1,66 @@
+# Resource utilization
+
+Implemented from [RFC #49](https://github.com/inclusionAI/sandboxd/issues/49).
+
+When `[plugin.node_resource]` is configured, the existing Unix HTTP endpoint `GET /resource` reports cached utilization alongside scheduler capacity. For example:
+
+```json
+{
+  "cpu": 32,
+  "mem": 68719476736,
+  "xpu": [],
+  "storage": 17179869184,
+  "features": ["storage-quota-v1"],
+  "utilization": {
+    "cpu": 0.92,
+    "memory": 0.87,
+    "pid": 0.88,
+    "fd": 0.22,
+    "disk": 0.75
+  }
+}
+```
+
+All five utilization fields are always present. Values are fractions from 0 to 1, clamped at 1 when consumption exceeds a limit. `null` means no usable observation; zero is a valid observation. CPU needs two samples, so it is initially `null`. The example values are illustrative.
+
+## Sampling
+
+The existing resource refresh loop samples every five seconds. HTTP requests read the latest cache and do not collect metrics. Sampling is read-only, works with either the Kubernetes or cgroup capacity provider, and does not require an additional configuration section.
+
+| Field | Sources and calculation |
+| --- | --- |
+| `cpu` | Maximum of host busy-time fraction and applicable finite cgroup quota utilization. Host busy time uses deltas from `/proc/stat`, excludes idle and iowait, and avoids counting guest time twice. Cgroup CPU time delta is divided by elapsed time and that cgroup's quota in cores. |
+| `memory` | Maximum of host `(MemTotal - MemAvailable) / MemTotal` and cgroup usage divided by a finite limit. Host fields come from `/proc/meminfo`. Cgroup usage includes charged cache. |
+| `pid` | Maximum `pids.current / pids.max` across the sampled cgroups. Counts tasks, including threads. An unlimited or unavailable limit supplies no ratio. |
+| `fd` | Maximum of `(allocated - unused) / maximum` from `/proc/sys/fs/file-nr` and the number of entries in `/proc/self/fd` divided by sandboxd's soft open-file limit from `/proc/self/limits`. System file handles and process descriptors are distinct constraints. |
+| `disk` | `1 - Bavail / Blocks` from the same `statfs(FilestoreDir)` snapshot as `storage`, before storage overcommit. Without a usable filestore snapshot this is `null`. |
+
+Cgroup sampling covers the daemon's own cgroup and visible ancestors. When sandboxd cgroup management is enabled, it also includes the configured sandbox root and its visible ancestors. Each ratio pairs usage and limit at the same cgroup. Discovery follows the visible mounts on each sample, supports cgroup v1 and v2, and does not enumerate individual sandbox cgroups.
+
+For v2, CPU uses `cpu.stat` and `cpu.max`, memory uses `memory.current` and `memory.max`, and tasks use `pids.current` and `pids.max`. For v1, CPU uses `cpuacct.usage` with `cpu.cfs_quota_us` / `cpu.cfs_period_us`, and memory uses `memory.usage_in_bytes` / `memory.limit_in_bytes`. Split v1 CPU and accounting mounts must have matching membership. Unlimited v1 memory sentinels are ignored.
+
+If one source is unavailable, other valid sources for that metric remain usable. A failed CPU read, counter reset, or changed quota restarts that source's delta baseline. An unavailable metric does not fail the resource response or change the module's liveness check. Existing scheduler-capacity caching remains intact when its provider fails.
+
+## Storage identity and coverage
+
+`disk` describes precisely the filesystem backing the reported `storage`. With loop-backed filestore, this is the mounted loop filesystem. With ordinary directory mode, it is the filesystem containing `FilestoreDir`. It measures physical space unavailable to ordinary allocations, including filesystem reservations. Changing the storage overcommit ratio changes advertised bytes but leaves disk occupancy unchanged.
+
+Metrics reflect only the procfs and cgroup hierarchy visible to sandboxd. CPU saturation caused solely by cpuset restrictions is not measured separately. Without a finite cgroup task limit, `pid` is `null`; it does not estimate exhaustion from the largest PID. FD reporting covers system file handles and sandboxd itself, not every control-plane process. Disk occupancy does not measure inode exhaustion, directory quota, or I/O pressure.
+
+## External scheduler use
+
+An existing external collector can read the whole utilization object from the same resource query. Treat observations as advisory protection signals. The scheduler owns thresholds, pressure reasons, and isolation/recovery counters; sandboxd does not mark a node unschedulable. Consumers must tolerate absent utilization from older sandboxd versions, `null` values, and future unknown fields.
+
+FunctionSystem integration is separate follow-up work. Its consumer can track specific CPU, memory, PID, FD, and disk pressures, require sustained high observations to enter isolation, and recover after fewer low observations. Recovery should clear only the relevant pressure while preserving lifecycle and manual isolation.
+
+## Validation
+
+The unit suite includes deterministic procfs fixtures, both cgroup versions, ancestor usage/limit pairing, CPU baseline changes, null/zero handling, and storage overcommit independence. Unix socket tests check the cached JSON contract. `TestModuleLiveUtilizationOverUnixSocket` exercises real Linux procfs and filestore statistics through the periodic sampler and HTTP endpoint, using a stub for the independent scheduler-capacity provider.
+
+```sh
+go test -race ./pkg/resourcemanager ./pkg/volumemanager
+go test -v ./pkg/resourcemanager -run TestModuleLiveUtilizationOverUnixSocket
+make check-fmt
+make vet
+make test
+```

internal/server/server.go

@@ -713,6 +713,13 @@ func NewSandboxService(root, configPath string) (result SandboxService, retErr e
 		if merr != nil {
 			return nil, fmt.Errorf("node-resource module init: %w", merr)
 		}
+		if !cfg.DisableCgroup && cfg.CgroupCacheSize > 0 {
+			root := cfg.CgroupRootName
+			if root == "" {
+				root = config.DefaultCgroupRoot
+			}
+			mod.SetSandboxCgroupRoot(root)
+		}
 		mod.SetXPUProvider(xpuMgr)
 		if serr := mod.Start(); serr != nil {
 			// NewModule already started the OTel collector's periodic-reader

pkg/resourcemanager/module.go

@@ -53,6 +53,8 @@ type Module struct {
 	ephemeralStorageAvailable uint64
 	ephemeralStorageReady     bool
 	transientMemory           map[string]int64
+	utilization               Utilization
+	utilizationSource         utilizationSource
 
 	lastRefresh atomic.Int64 // unix-nano of most recent successful refresh
 	listener    net.Listener
@@ -67,11 +69,12 @@ type Module struct {
 // resourceInfo is the JSON payload served by the /resource endpoint. CPU is
 // reported in scheduler cores, while memory and writable storage use bytes.
 type resourceInfo struct {
-	Cpu      int64                 `json:"cpu"`
-	Mem      int64                 `json:"mem"`
-	Xpu      []xpumanager.Resource `json:"xpu"`
-	Storage  *uint64               `json:"storage,omitempty"`
-	Features []string              `json:"features"`
+	Cpu         int64                 `json:"cpu"`
+	Mem         int64                 `json:"mem"`
+	Xpu         []xpumanager.Resource `json:"xpu"`
+	Storage     *uint64               `json:"storage,omitempty"`
+	Features    []string              `json:"features"`
+	Utilization Utilization           `json:"utilization"`
 }
 
 type xpuProvider interface {
@@ -82,6 +85,12 @@ type ephemeralStorageProvider interface {
 	EphemeralStorageCapacity() (capacityBytes, allocatableBytes uint64, err error)
 }
 
+// Snapshot providers return physical occupancy alongside logical storage from
+// one filesystem read. Capacity-only providers remain supported with disk=null.
+type ephemeralStorageSnapshotProvider interface {
+	EphemeralStorageSnapshot() (capacityBytes, allocatableBytes uint64, utilization *float64, err error)
+}
+
 const storageQuotaFeature = "storage-quota-v1"
 
 // NewModule constructs the configured node-resource module. sockPath is the
@@ -93,10 +102,11 @@ func NewModule(sockPath, provider string) (*Module, error) {
 	}
 
 	m := &Module{
-		nodeResource:    nrm,
-		sockPath:        sockPath,
-		stopCh:          make(chan struct{}),
-		transientMemory: make(map[string]int64),
+		nodeResource:      nrm,
+		utilizationSource: newUtilizationSampler(""),
+		sockPath:          sockPath,
+		stopCh:            make(chan struct{}),
+		transientMemory:   make(map[string]int64),
 	}
 
 	// OTLP metrics push is best-effort: a missing collector at startup must
@@ -115,6 +125,12 @@ func NewModule(sockPath, provider string) (*Module, error) {
 	return m, nil
 }
 
+// SetSandboxCgroupRoot includes the managed sandbox root in utilization
+// sampling. Call before Start; an empty root samples only the daemon hierarchy.
+func (m *Module) SetSandboxCgroupRoot(root string) {
+	m.utilizationSource = newUtilizationSampler(root)
+}
+
 // SetSandboxMetricsSource connects the node metrics collector to sandbox
 // lifecycle state once the sandbox manager is ready.
 func (m *Module) SetSandboxMetricsSource(source SandboxMetricsSource) {
@@ -174,10 +190,11 @@ func (m *Module) Start() error {
 	mux.HandleFunc("/resource", func(w http.ResponseWriter, r *http.Request) {
 		m.mu.RLock()
 		info := resourceInfo{
-			Cpu:      m.availCpu,
-			Mem:      availableAfterTransientReservations(m.availMem, m.transientMemory),
-			Xpu:      []xpumanager.Resource{},
-			Features: []string{},
+			Cpu:         m.availCpu,
+			Mem:         availableAfterTransientReservations(m.availMem, m.transientMemory),
+			Xpu:         []xpumanager.Resource{},
+			Features:    []string{},
+			Utilization: m.utilization,
 		}
 		if m.xpu != nil {
 			info.Xpu = m.xpu.Resources()
@@ -327,6 +344,8 @@ func (m *Module) refreshLoop() {
 }
 
 func (m *Module) refreshOnce() {
+	m.refreshUtilization()
+	m.refreshEphemeralStorage()
 	cpu, mem, err := m.nodeResource.GetAvailableResource()
 	if err != nil {
 		logrus.Errorf("resourcemanager: refresh failed: %v", err)
@@ -341,22 +360,45 @@ func (m *Module) refreshOnce() {
 	m.mu.Unlock()
 	m.lastRefresh.Store(time.Now().UnixNano())
 	logrus.Debugf("resourcemanager: avail cpu=%d cores mem=%d bytes", cpu, mem)
-	m.refreshEphemeralStorage()
+}
+
+func (m *Module) refreshUtilization() {
+	if m.utilizationSource == nil {
+		return
+	}
+	observation := m.utilizationSource.Sample()
+	m.mu.Lock()
+	observation.Disk = m.utilization.Disk
+	m.utilization = observation
+	m.mu.Unlock()
 }
 
 func (m *Module) refreshEphemeralStorage() {
 	m.mu.RLock()
 	provider := m.ephemeralStorage
 	m.mu.RUnlock()
 	if provider == nil {
+		m.mu.Lock()
+		m.utilization.Disk = nil
+		m.mu.Unlock()
 		return
 	}
-	capacity, allocatable, err := provider.EphemeralStorageCapacity()
+	var capacity, allocatable uint64
+	var disk *float64
+	var err error
+	if snapshot, ok := provider.(ephemeralStorageSnapshotProvider); ok {
+		capacity, allocatable, disk, err = snapshot.EphemeralStorageSnapshot()
+	} else {
+		capacity, allocatable, err = provider.EphemeralStorageCapacity()
+	}
+	m.mu.Lock()
 	if err != nil {
+		m.utilization.Disk = nil
+		m.mu.Unlock()
 		logrus.Errorf("resourcemanager: ephemeral storage refresh failed: %v", err)
 		return
 	}
-	m.mu.Lock()
+	m.utilization.Disk = disk
 	m.ephemeralStorageCapacity = capacity
 	m.ephemeralStorageAvailable = allocatable
 	m.ephemeralStorageReady = true

pkg/resourcemanager/module_test.go

@@ -17,6 +17,7 @@ package resourcemanager
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"io"
 	"net"
 	"net/http"
@@ -25,6 +26,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/inclusionAI/sandboxd/pkg/volumemanager"
 	"github.com/inclusionAI/sandboxd/pkg/xpumanager"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -114,9 +116,10 @@ func TestModuleServesAvailableResourceOverUnixSocket(t *testing.T) {
 	resourceManager := &stubNodeResourceManager{cpu: 2500, mem: 5 << 30}
 	sockPath := filepath.Join(t.TempDir(), "resource.sock")
 	m := &Module{
-		nodeResource: resourceManager,
-		sockPath:     sockPath,
-		stopCh:       make(chan struct{}),
+		nodeResource:      resourceManager,
+		utilizationSource: fixedUtilizationSource{Utilization{CPU: utilizationRatio(0, 1), Memory: utilizationRatio(3, 4)}},
+		sockPath:          sockPath,
+		stopCh:            make(chan struct{}),
 		xpu: stubXPUProvider{resources: []xpumanager.Resource{{
 			Type:         "gpu",
 			ProductModel: "l20",
@@ -152,6 +155,13 @@ func TestModuleServesAvailableResourceOverUnixSocket(t *testing.T) {
 		if err := json.Unmarshal(body, &fields); err != nil {
 			return false
 		}
+		var utilization map[string]*float64
+		if err := json.Unmarshal(fields["utilization"], &utilization); err != nil {
+			return false
+		}
+		if len(utilization) != 5 || utilization["cpu"] == nil || *utilization["cpu"] != 0 || utilization["memory"] == nil || *utilization["memory"] != .75 || utilization["pid"] != nil || utilization["fd"] != nil || utilization["disk"] != nil {
+			return false
+		}
 		if _, ok := fields["storage"]; !ok {
 			return false
 		}
@@ -175,3 +185,116 @@ func TestModuleServesAvailableResourceOverUnixSocket(t *testing.T) {
 // Module must satisfy the metrics CapacityProvider so the collector can read
 // the socket-aligned availability figure.
 var _ CapacityProvider = (*Module)(nil)
+
+type fixedUtilizationSource struct{ observation Utilization }
+
+func (s fixedUtilizationSource) Sample() Utilization { return s.observation }
+
+type failedNodeResourceManager struct{}
+
+func (failedNodeResourceManager) GetAvailableResource() (int64, int64, error) {
+	return 0, 0, errors.New("capacity unavailable")
+}
+func (failedNodeResourceManager) Stop() {}
+
+type snapshotStorageProvider struct {
+	stubEphemeralStorageProvider
+	disk *float64
+	err  error
+}
+
+func (s snapshotStorageProvider) EphemeralStorageCapacity() (uint64, uint64, error) {
+	panic("snapshot provider must use a single snapshot call")
+}
+func (s snapshotStorageProvider) EphemeralStorageSnapshot() (uint64, uint64, *float64, error) {
+	return s.capacity, s.allocatable, s.disk, s.err
+}
+
+func TestUtilizationRefreshIndependentOfCapacity(t *testing.T) {
+	m := &Module{nodeResource: failedNodeResourceManager{}, utilizationSource: fixedUtilizationSource{Utilization{CPU: utilizationRatio(9, 10)}}}
+	m.SetEphemeralStorageProvider(snapshotStorageProvider{stubEphemeralStorageProvider: stubEphemeralStorageProvider{capacity: 200, allocatable: 50}, disk: utilizationRatio(3, 4)})
+	m.refreshOnce()
+	requireUtilization(t, .9, m.utilization.CPU)
+	requireUtilization(t, .75, m.utilization.Disk)
+	require.False(t, m.Healthy()) // utilization does not redefine module liveness
+	m.utilizationSource = fixedUtilizationSource{}
+	m.refreshOnce()
+	require.Nil(t, m.utilization.CPU)
+	requireUtilization(t, .75, m.utilization.Disk)
+	m.SetEphemeralStorageProvider(snapshotStorageProvider{err: errors.New("statfs unavailable")})
+	require.Nil(t, m.utilization.Disk)
+	require.Equal(t, uint64(50), m.ephemeralStorageAvailable) // existing capacity caching
+}
+
+func TestModuleServesCachedUtilization(t *testing.T) {
+	sockPath := filepath.Join(t.TempDir(), "resource.sock")
+	calls := atomic.Int32{}
+	source := utilizationSourceFunc(func() Utilization { calls.Add(1); return Utilization{PID: utilizationRatio(4, 5)} })
+	m := &Module{nodeResource: &stubNodeResourceManager{}, sockPath: sockPath, stopCh: make(chan struct{}), utilizationSource: source}
+	require.NoError(t, m.Start())
+	defer m.Stop()
+	require.Eventually(t, func() bool {
+		m.mu.RLock()
+		defer m.mu.RUnlock()
+		return m.utilization.PID != nil
+	}, time.Second, time.Millisecond)
+	transport := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+		return (&net.Dialer{}).DialContext(ctx, "unix", sockPath)
+	}}
+	defer transport.CloseIdleConnections()
+	client := &http.Client{Transport: transport, Timeout: time.Second}
+	for i := 0; i < 3; i++ {
+		resp, err := client.Get("http://sandboxd/resource")
+		require.NoError(t, err)
+		var info resourceInfo
+		err = json.NewDecoder(resp.Body).Decode(&info)
+		resp.Body.Close()
+		require.NoError(t, err)
+		requireUtilization(t, .8, info.Utilization.PID)
+	}
+	require.Equal(t, int32(1), calls.Load())
+}
+
+type utilizationSourceFunc func() Utilization
+
+func (f utilizationSourceFunc) Sample() Utilization { return f() }
+
+// Exercise real Linux procfs/cgroup discovery and filestore statfs through the
+// periodic sampler and the public Unix HTTP endpoint. Capacity is independent.
+func TestModuleLiveUtilizationOverUnixSocket(t *testing.T) {
+	sockPath := filepath.Join(t.TempDir(), "resource.sock")
+	m := &Module{nodeResource: &stubNodeResourceManager{cpu: 2000, mem: 1024}, sockPath: sockPath, stopCh: make(chan struct{}), utilizationSource: newUtilizationSampler("")}
+	m.SetEphemeralStorageProvider(volumemanager.NewModule(t.TempDir(), "", false, 2))
+	require.NoError(t, m.Start())
+	defer m.Stop()
+	transport := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+		return (&net.Dialer{}).DialContext(ctx, "unix", sockPath)
+	}}
+	defer transport.CloseIdleConnections()
+	client := &http.Client{Transport: transport, Timeout: time.Second}
+	var info resourceInfo
+	require.Eventually(t, func() bool {
+		resp, err := client.Get("http://sandboxd/resource")
+		if err != nil {
+			return false
+		}
+		defer resp.Body.Close()
+		if json.NewDecoder(resp.Body).Decode(&info) != nil {
+			return false
+		}
+		return info.Utilization.CPU != nil
+	}, 10*time.Second, 100*time.Millisecond)
+	require.NotNil(t, info.Utilization.Memory)
+	require.NotNil(t, info.Utilization.FD)
+	require.NotNil(t, info.Utilization.Disk)
+	require.NotNil(t, info.Storage)
+	for _, value := range []*float64{info.Utilization.CPU, info.Utilization.Memory, info.Utilization.PID, info.Utilization.FD, info.Utilization.Disk} {
+		if value != nil {
+			require.GreaterOrEqual(t, *value, 0.0)
+			require.LessOrEqual(t, *value, 1.0)
+		}
+	}
+	body, err := json.Marshal(info)
+	require.NoError(t, err)
+	t.Logf("live /resource response: %s", body)
+}