diff --git a/README.md b/README.md index f22f9c2..7cdf712 100644 --- a/README.md +++ b/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. diff --git a/doc/resource-utilization.md b/doc/resource-utilization.md new file mode 100644 index 0000000..2887b04 --- /dev/null +++ b/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 +``` diff --git a/internal/server/server.go b/internal/server/server.go index cc16c2a..dd250c8 100644 --- a/internal/server/server.go +++ b/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 diff --git a/pkg/resourcemanager/module.go b/pkg/resourcemanager/module.go index 033aa12..529f17f 100644 --- a/pkg/resourcemanager/module.go +++ b/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,7 +360,17 @@ 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() { @@ -349,14 +378,27 @@ func (m *Module) refreshEphemeralStorage() { 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 diff --git a/pkg/resourcemanager/module_test.go b/pkg/resourcemanager/module_test.go index d051b21..3ffe002 100644 --- a/pkg/resourcemanager/module_test.go +++ b/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) +} diff --git a/pkg/resourcemanager/utilization.go b/pkg/resourcemanager/utilization.go new file mode 100644 index 0000000..c2bac75 --- /dev/null +++ b/pkg/resourcemanager/utilization.go @@ -0,0 +1,351 @@ +// Copyright (c) 2026 Ant Group Corporation. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package resourcemanager + +import ( + "bufio" + "math" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/containerd/cgroups/v3" + "github.com/moby/sys/mountinfo" +) + +// Utilization contains advisory fractions, independent of scheduler capacity. +// A nil field is encoded as null, distinguishing unavailable data from zero. +type Utilization struct { + CPU *float64 `json:"cpu"` + Memory *float64 `json:"memory"` + PID *float64 `json:"pid"` + FD *float64 `json:"fd"` + Disk *float64 `json:"disk"` +} + +type utilizationSource interface{ Sample() Utilization } + +type utilizationGroup struct { + kind string + path string + accounting string // CPU accounting may have a separate v1 mount. + unified bool +} + +type cpuObservation struct { + usage uint64 + idle uint64 + at time.Time + limit float64 +} + +type utilizationSampler struct { + proc string + now func() time.Time + groups func() []utilizationGroup + previous map[string]cpuObservation +} + +func newUtilizationSampler(sandboxRoot string) *utilizationSampler { + return &utilizationSampler{ + proc: "/proc", now: time.Now, + groups: func() []utilizationGroup { return discoverUtilizationGroups(sandboxRoot) }, + previous: make(map[string]cpuObservation), + } +} + +// Sample performs bounded, read-only work: no per-sandbox or process-tree scan. +// Only this sampler's refresh-loop goroutine calls Sample. +func (s *utilizationSampler) Sample() Utilization { + now := s.now() + next := make(map[string]cpuObservation) + result := Utilization{Memory: hostMemoryUtilization(s.proc), FD: fdUtilization(s.proc)} + if current, ok := hostCPUObservation(s.proc); ok { + next["host"] = current + if previous, exists := s.previous["host"]; exists && current.usage > previous.usage && current.idle >= previous.idle { + total, idle := current.usage-previous.usage, current.idle-previous.idle + if idle <= total { + result.CPU = utilizationRatio(float64(total-idle), float64(total)) + } + } + } + for _, group := range s.groups() { + switch group.kind { + case "cpu": + current, ok := cgroupCPUObservation(group) + if !ok { + continue + } + current.at = now + key := group.path + next[key] = current + previous, exists := s.previous[key] + elapsed := now.Sub(previous.at).Seconds() + if exists && current.limit == previous.limit && current.usage >= previous.usage && elapsed > 0 { + ratio := utilizationRatio(float64(current.usage-previous.usage)/float64(time.Second), elapsed*current.limit) + result.CPU = maximumUtilization(result.CPU, ratio) + } + case "memory": + usage, limit := "memory.usage_in_bytes", "memory.limit_in_bytes" + if group.unified { + usage, limit = "memory.current", "memory.max" + } + result.Memory = maximumUtilization(result.Memory, cgroupUtilization(group.path, usage, limit, !group.unified)) + case "pids": + result.PID = maximumUtilization(result.PID, cgroupUtilization(group.path, "pids.current", "pids.max", false)) + } + } + s.previous = next + return result +} + +func utilizationRatio(usage, limit float64) *float64 { + if math.IsNaN(usage) || math.IsInf(usage, 0) || math.IsNaN(limit) || math.IsInf(limit, 0) || usage < 0 || limit <= 0 { + return nil + } + value := math.Min(usage/limit, 1) + return &value +} + +func maximumUtilization(a, b *float64) *float64 { + if a == nil || (b != nil && *b > *a) { + return b + } + return a +} + +func readUintFile(path string) (uint64, bool) { + data, err := os.ReadFile(path) + if err != nil { + return 0, false + } + value, err := strconv.ParseUint(strings.TrimSpace(string(data)), 10, 64) + return value, err == nil +} + +func cgroupUtilization(path, usageFile, limitFile string, v1Memory bool) *float64 { + limit, ok := readUintFile(filepath.Join(path, limitFile)) + // v1 represents unlimited memory with a page-aligned LONG_MAX sentinel. + if !ok || limit == 0 || (v1Memory && limit >= 1<<60) { + return nil + } + usage, ok := readUintFile(filepath.Join(path, usageFile)) + if !ok { + return nil + } + return utilizationRatio(float64(usage), float64(limit)) +} + +func hostCPUObservation(proc string) (cpuObservation, bool) { + data, err := os.ReadFile(filepath.Join(proc, "stat")) + if err != nil { + return cpuObservation{}, false + } + fields := strings.Fields(strings.SplitN(string(data), "\n", 2)[0]) + if len(fields) < 5 || fields[0] != "cpu" { + return cpuObservation{}, false + } + result := cpuObservation{} + // user, nice, system, idle, iowait, irq, softirq, steal. guest is already + // included in user/nice, so additional fields must not be added again. + for i := 1; i < len(fields) && i <= 8; i++ { + value, err := strconv.ParseUint(fields[i], 10, 64) + if err != nil || value > math.MaxUint64-result.usage { + return cpuObservation{}, false + } + result.usage += value + if i == 4 || i == 5 { + result.idle += value + } + } + return result, true +} + +func hostMemoryUtilization(proc string) *float64 { + data, err := os.ReadFile(filepath.Join(proc, "meminfo")) + if err != nil { + return nil + } + var total, available uint64 + var haveTotal, haveAvailable bool + for line := range strings.SplitSeq(string(data), "\n") { + fields := strings.Fields(line) + if len(fields) != 3 || fields[2] != "kB" { + continue + } + value, err := strconv.ParseUint(fields[1], 10, 64) + if err != nil { + continue + } + switch fields[0] { + case "MemTotal:": + total, haveTotal = value, true + case "MemAvailable:": + available, haveAvailable = value, true + } + } + if !haveTotal || !haveAvailable || available > total { + return nil + } + return utilizationRatio(float64(total-available), float64(total)) +} + +func fdUtilization(proc string) *float64 { + var result *float64 + if data, err := os.ReadFile(filepath.Join(proc, "sys/fs/file-nr")); err == nil { + fields := strings.Fields(string(data)) + if len(fields) == 3 { + allocated, e1 := strconv.ParseUint(fields[0], 10, 64) + unused, e2 := strconv.ParseUint(fields[1], 10, 64) + limit, e3 := strconv.ParseUint(fields[2], 10, 64) + if e1 == nil && e2 == nil && e3 == nil && unused <= allocated { + result = utilizationRatio(float64(allocated-unused), float64(limit)) + } + } + } + data, err := os.ReadFile(filepath.Join(proc, "self/limits")) + if err != nil { + return result + } + for line := range strings.SplitSeq(string(data), "\n") { + fields := strings.Fields(line) + if len(fields) != 6 || strings.Join(fields[:3], " ") != "Max open files" { + continue + } + limit, err := strconv.ParseUint(fields[3], 10, 64) + if err != nil { + break + } + entries, err := os.ReadDir(filepath.Join(proc, "self/fd")) + if err == nil { + result = maximumUtilization(result, utilizationRatio(float64(len(entries)), float64(limit))) + } + break + } + return result +} + +func cgroupCPUObservation(group utilizationGroup) (cpuObservation, bool) { + var milli int64 + var limited bool + var err error + var usage uint64 + var ok bool + if group.unified { + milli, limited, err = readV2CPUQuota(filepath.Join(group.path, "cpu.max")) + if err != nil || !limited { + return cpuObservation{}, false + } + data, readErr := os.ReadFile(filepath.Join(group.path, "cpu.stat")) + if readErr != nil { + return cpuObservation{}, false + } + scanner := bufio.NewScanner(strings.NewReader(string(data))) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) == 2 && fields[0] == "usage_usec" { + value, parseErr := strconv.ParseUint(fields[1], 10, 64) + if parseErr == nil && value <= math.MaxUint64/uint64(time.Microsecond) { + usage, ok = value*uint64(time.Microsecond), true + } + break + } + } + } else { + milli, limited, err = readV1CPUQuota(group.path) + if err != nil || !limited { + return cpuObservation{}, false + } + usage, ok = readUintFile(filepath.Join(group.accounting, "cpuacct.usage")) + } + return cpuObservation{usage: usage, limit: float64(milli) / 1000}, ok +} + +func discoverUtilizationGroups(sandboxRoot string) []utilizationGroup { + data, err := os.Open(procSelfCgroupPath) + if err != nil { + return nil + } + defer data.Close() + self, err := parseSelfCgroups(data) + if err != nil { + return nil + } + mounts, err := mountinfo.GetMounts(mountinfo.FSTypeFilter("cgroup", "cgroup2")) + if err != nil { + return nil + } + return utilizationGroups(mounts, self, cgroups.Mode() == cgroups.Unified, sandboxRoot) +} + +func utilizationGroups(mounts []*mountinfo.Info, self selfCgroups, unified bool, sandboxRoot string) []utilizationGroup { + var result []utilizationGroup + seen := make(map[string]bool) + for _, kind := range []string{"cpu", "memory", "pids"} { + fsType, controller, membership := "cgroup", kind, self.controller[kind] + if unified { + fsType, controller, membership = "cgroup2", "", self.unified + } + if membership == "" { + continue + } + hierarchy, err := findHierarchy(mounts, fsType, controller, membership) + if err != nil { + continue + } + paths := hierarchyAncestors(hierarchy) + if sandboxRoot != "" { + root := filepath.Join(hierarchy.mountpoint, filepath.Clean("/"+sandboxRoot)) + paths = append(paths, hierarchyAncestors(cgroupHierarchy{mountpoint: hierarchy.mountpoint, current: root})...) + } + for _, path := range paths { + key := kind + ":" + path + if seen[key] { + continue + } + seen[key] = true + group := utilizationGroup{kind: kind, path: path, unified: unified} + if kind == "cpu" && !unified { + // A split cpuacct hierarchy is usable only when membership matches. + // Resolve each ancestor's logical path to preserve usage/limit scope. + if self.controller["cpuacct"] != membership { + continue + } + for _, mount := range mounts { + if mount.Mountpoint != hierarchy.mountpoint { + continue + } + relative, relErr := filepath.Rel(hierarchy.mountpoint, path) + if relErr != nil { + continue + } + logical := filepath.Join(mount.Root, relative) + accounting, acctErr := findHierarchy(mounts, "cgroup", "cpuacct", logical) + if acctErr == nil { + group.accounting = accounting.current + } + break + } + if group.accounting == "" { + continue + } + } + result = append(result, group) + } + } + return result +} diff --git a/pkg/resourcemanager/utilization_test.go b/pkg/resourcemanager/utilization_test.go new file mode 100644 index 0000000..a842f95 --- /dev/null +++ b/pkg/resourcemanager/utilization_test.go @@ -0,0 +1,188 @@ +// Copyright (c) 2026 Ant Group Corporation. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package resourcemanager + +import ( + "math" + "os" + "path/filepath" + "testing" + "time" + + "github.com/moby/sys/mountinfo" + "github.com/stretchr/testify/require" +) + +func writeUtilizationFixture(t *testing.T, root, name, content string) { + t.Helper() + path := filepath.Join(root, name) + require.NoError(t, os.MkdirAll(filepath.Dir(path), 0755)) + require.NoError(t, os.WriteFile(path, []byte(content), 0644)) +} + +func requireUtilization(t *testing.T, want float64, got *float64) { + t.Helper() + require.NotNil(t, got) + require.InDelta(t, want, *got, 0.000001) +} + +func TestHostUtilizationSampling(t *testing.T) { + proc := t.TempDir() + writeUtilizationFixture(t, proc, "stat", "cpu 10 0 10 80 0 0 0 0 100 100\n") + writeUtilizationFixture(t, proc, "meminfo", "MemTotal: 1000 kB\nMemAvailable: 250 kB\nMemFree: 10 kB\n") + s := &utilizationSampler{proc: proc, now: time.Now, groups: func() []utilizationGroup { return nil }} + first := s.Sample() + require.Nil(t, first.CPU) + requireUtilization(t, .75, first.Memory) + require.Nil(t, first.PID) + require.Nil(t, first.FD) + require.Nil(t, first.Disk) + writeUtilizationFixture(t, proc, "stat", "cpu 40 0 20 130 10 0 0 0 1000 1000\n") + requireUtilization(t, .4, s.Sample().CPU) // delta total=100, idle+iowait=60 + writeUtilizationFixture(t, proc, "stat", "cpu 1 0 1 8 0 0 0 0\n") + require.Nil(t, s.Sample().CPU) // counter reset + require.NoError(t, os.Remove(filepath.Join(proc, "stat"))) + require.Nil(t, s.Sample().CPU) + writeUtilizationFixture(t, proc, "stat", "cpu 50 0 20 130 10 0 0 0\n") + require.Nil(t, s.Sample().CPU) // failed sample broke the delta baseline + writeUtilizationFixture(t, proc, "meminfo", "MemTotal: 1000 kB\n") + require.Nil(t, s.Sample().Memory) // do not substitute MemFree + writeUtilizationFixture(t, proc, "meminfo", "MemTotal: 1000 kB\nMemAvailable: 1000 kB\n") + requireUtilization(t, 0, s.Sample().Memory) +} + +func TestUtilizationRatioBounds(t *testing.T) { + for _, limit := range []float64{0, -1, math.NaN(), math.Inf(1)} { + require.Nil(t, utilizationRatio(1, limit)) + } + for _, usage := range []float64{-1, math.NaN(), math.Inf(1)} { + require.Nil(t, utilizationRatio(usage, 1)) + } + requireUtilization(t, 1, utilizationRatio(200, 100)) + requireUtilization(t, 0, utilizationRatio(0, 100)) +} + +func TestFDUtilization(t *testing.T) { + proc := t.TempDir() + writeUtilizationFixture(t, proc, "sys/fs/file-nr", "80 20 100\n") + requireUtilization(t, .6, fdUtilization(proc)) + writeUtilizationFixture(t, proc, "self/limits", "Limit Soft Limit Hard Limit Units\nMax open files 10 100 files\n") + for _, fd := range []string{"0", "1", "2", "3", "4", "5", "6", "7"} { + writeUtilizationFixture(t, proc, "self/fd/"+fd, "") + } + requireUtilization(t, .8, fdUtilization(proc)) + writeUtilizationFixture(t, proc, "self/limits", "Max open files unlimited unlimited files\n") + requireUtilization(t, .6, fdUtilization(proc)) + writeUtilizationFixture(t, proc, "sys/fs/file-nr", "bad data\n") + require.Nil(t, fdUtilization(proc)) +} + +func TestCgroupUtilizationV1AndV2(t *testing.T) { + for _, unified := range []bool{false, true} { + name := "v1" + if unified { + name = "v2" + } + t.Run(name, func(t *testing.T) { + root := t.TempDir() + memoryUsage, memoryLimit := "memory.usage_in_bytes", "memory.limit_in_bytes" + cpuUsage := "cpuacct.usage" + if unified { + memoryUsage, memoryLimit, cpuUsage = "memory.current", "memory.max", "cpu.stat" + writeUtilizationFixture(t, root, "cpu.max", "200000 100000") + writeUtilizationFixture(t, root, cpuUsage, "usage_usec 1000000\n") + } else { + writeUtilizationFixture(t, root, "cpu.cfs_quota_us", "200000") + writeUtilizationFixture(t, root, "cpu.cfs_period_us", "100000") + writeUtilizationFixture(t, root, cpuUsage, "1000000000") + } + writeUtilizationFixture(t, root, memoryUsage, "80") + writeUtilizationFixture(t, root, memoryLimit, "100") + writeUtilizationFixture(t, root, "pids.current", "95") + writeUtilizationFixture(t, root, "pids.max", "100") + now := time.Unix(100, 0) + s := &utilizationSampler{proc: t.TempDir(), now: func() time.Time { return now }, groups: func() []utilizationGroup { + return []utilizationGroup{{kind: "cpu", path: root, accounting: root, unified: unified}, {kind: "memory", path: root, unified: unified}, {kind: "pids", path: root, unified: unified}} + }} + first := s.Sample() + require.Nil(t, first.CPU) + requireUtilization(t, .8, first.Memory) + requireUtilization(t, .95, first.PID) + now = now.Add(5 * time.Second) + if unified { + writeUtilizationFixture(t, root, cpuUsage, "usage_usec 10000000\n") + } else { + writeUtilizationFixture(t, root, cpuUsage, "10000000000") + } + requireUtilization(t, .9, s.Sample().CPU) + writeUtilizationFixture(t, root, "pids.max", "max") + unlimitedMemory := "9223372036854771712" + if unified { + unlimitedMemory = "max" + } + writeUtilizationFixture(t, root, memoryLimit, unlimitedMemory) + sample := s.Sample() + require.Nil(t, sample.PID) + require.Nil(t, sample.Memory) + // Changing the denominator must start a new CPU baseline. + if unified { + writeUtilizationFixture(t, root, "cpu.max", "100000 100000") + } else { + writeUtilizationFixture(t, root, "cpu.cfs_quota_us", "100000") + } + now = now.Add(5 * time.Second) + require.Nil(t, s.Sample().CPU) + }) + } +} + +func TestCgroupAncestorUsageAndLimitStayPaired(t *testing.T) { + root := t.TempDir() + child := filepath.Join(root, "sandbox") + writeUtilizationFixture(t, root, "pids.current", "90") + writeUtilizationFixture(t, root, "pids.max", "100") + writeUtilizationFixture(t, child, "pids.current", "20") + writeUtilizationFixture(t, child, "pids.max", "50") + groups := utilizationGroups([]*mountinfo.Info{{Mountpoint: root, Root: "/", FSType: "cgroup2"}}, selfCgroups{unified: "/daemon"}, true, "sandbox") + s := &utilizationSampler{proc: t.TempDir(), now: time.Now, groups: func() []utilizationGroup { return groups }} + requireUtilization(t, .9, s.Sample().PID) // ancestor 90/100, not child 20/100 + require.NoError(t, os.Remove(filepath.Join(root, "pids.current"))) + requireUtilization(t, .4, s.Sample().PID) // valid child survives unavailable ancestor +} + +func TestUtilizationCgroupDiscovery(t *testing.T) { + mounts := []*mountinfo.Info{ + {Mountpoint: "/cg/cpu", Root: "/tenant", FSType: "cgroup", VFSOptions: "rw,cpu"}, + {Mountpoint: "/cg/cpuacct", Root: "/tenant", FSType: "cgroup", VFSOptions: "rw,cpuacct"}, + {Mountpoint: "/cg/pids", Root: "/tenant", FSType: "cgroup", VFSOptions: "rw,pids"}, + } + self := selfCgroups{controller: map[string]string{"cpu": "/tenant/daemon", "cpuacct": "/tenant/daemon", "pids": "/tenant/daemon"}} + groups := utilizationGroups(mounts, self, false, "sandbox") + require.Contains(t, groups, utilizationGroup{kind: "cpu", path: "/cg/cpu/sandbox", accounting: "/cg/cpuacct/sandbox"}) + require.Contains(t, groups, utilizationGroup{kind: "pids", path: "/cg/pids/daemon"}) + require.Contains(t, groups, utilizationGroup{kind: "pids", path: "/cg/pids"}) + unique := make(map[string]bool) + for _, group := range groups { + key := group.kind + ":" + group.path + require.False(t, unique[key]) + unique[key] = true + require.NotEqual(t, "/cg", group.path) + } + self.controller["cpuacct"] = "/different" + for _, group := range utilizationGroups(mounts, self, false, "") { + require.NotEqual(t, "cpu", group.kind) + } + require.Empty(t, utilizationGroups(nil, self, false, "sandbox")) +} diff --git a/pkg/volumemanager/module.go b/pkg/volumemanager/module.go index aa9f052..324c1ba 100644 --- a/pkg/volumemanager/module.go +++ b/pkg/volumemanager/module.go @@ -124,28 +124,45 @@ func (m *Module) Healthy() bool { return m.healthy.Load() } // filestore. The overcommit ratio is applied exactly once here, at the boundary // between physical filesystem statistics and scheduler-visible storage. func (m *Module) EphemeralStorageCapacity() (uint64, uint64, error) { + capacity, available, _, err := m.EphemeralStorageSnapshot() + return capacity, available, err +} + +// EphemeralStorageSnapshot reports logical storage and physical occupancy from +// the same filestore statfs call. Overcommit affects bytes, never occupancy. +func (m *Module) EphemeralStorageSnapshot() (uint64, uint64, *float64, error) { if m.FilestoreDir == "" { - return 0, 0, fmt.Errorf("filestore_dir is not configured") + return 0, 0, nil, fmt.Errorf("filestore_dir is not configured") } var stat syscall.Statfs_t if err := syscall.Statfs(m.FilestoreDir, &stat); err != nil { - return 0, 0, fmt.Errorf("statfs %s: %w", m.FilestoreDir, err) + return 0, 0, nil, fmt.Errorf("statfs %s: %w", m.FilestoreDir, err) } - if stat.Bsize <= 0 { - return 0, 0, fmt.Errorf("statfs %s returned invalid block size %d", m.FilestoreDir, stat.Bsize) + return ephemeralStorageSnapshot(stat, m.overcommitRatio) +} + +func ephemeralStorageSnapshot(stat syscall.Statfs_t, ratio float64) (uint64, uint64, *float64, error) { + if stat.Bsize <= 0 || stat.Bavail > stat.Blocks { + return 0, 0, nil, fmt.Errorf("invalid filestore statfs: block size=%d blocks=%d available=%d", stat.Bsize, stat.Blocks, stat.Bavail) } blockSize := uint64(stat.Bsize) - physicalCapacity := stat.Blocks * blockSize - physicalAvailable := stat.Bavail * blockSize - capacity, err := scaleStorageBytes(physicalCapacity, m.overcommitRatio) + if stat.Blocks > math.MaxUint64/blockSize { + return 0, 0, nil, fmt.Errorf("filestore capacity overflows bytes") + } + capacity, err := scaleStorageBytes(stat.Blocks*blockSize, ratio) if err != nil { - return 0, 0, fmt.Errorf("scale filestore capacity: %w", err) + return 0, 0, nil, fmt.Errorf("scale filestore capacity: %w", err) } - available, err := scaleStorageBytes(physicalAvailable, m.overcommitRatio) + available, err := scaleStorageBytes(stat.Bavail*blockSize, ratio) if err != nil { - return 0, 0, fmt.Errorf("scale filestore available bytes: %w", err) + return 0, 0, nil, fmt.Errorf("scale filestore available bytes: %w", err) + } + var utilization *float64 + if stat.Blocks > 0 { + used := 1 - float64(stat.Bavail)/float64(stat.Blocks) + utilization = &used } - return capacity, available, nil + return capacity, available, utilization, nil } func scaleStorageBytes(physicalBytes uint64, ratio float64) (uint64, error) { diff --git a/pkg/volumemanager/module_test.go b/pkg/volumemanager/module_test.go index 750aedf..7afeafb 100644 --- a/pkg/volumemanager/module_test.go +++ b/pkg/volumemanager/module_test.go @@ -19,6 +19,7 @@ import ( "math" "os" "path/filepath" + "syscall" "testing" "github.com/inclusionAI/sandboxd/pkg/loopdevice" @@ -145,3 +146,42 @@ func TestScaleStorageBytes(t *testing.T) { }) } } + +func TestEphemeralStorageSnapshotUsesPhysicalOccupancy(t *testing.T) { + stat := syscall.Statfs_t{Bsize: 4096, Blocks: 100, Bavail: 25} + for _, ratio := range []float64{1, 1.5, 2} { + capacity, available, disk, err := ephemeralStorageSnapshot(stat, ratio) + if err != nil { + t.Fatal(err) + } + if capacity != uint64(409600*ratio) || available != uint64(102400*ratio) || disk == nil || *disk != .75 { + t.Fatalf("ratio=%v capacity=%d available=%d disk=%v", ratio, capacity, available, disk) + } + } + _, _, disk, err := ephemeralStorageSnapshot(syscall.Statfs_t{Bsize: 4096}, 1) + if err != nil || disk != nil { + t.Fatalf("empty filesystem: disk=%v err=%v", disk, err) + } + for _, invalid := range []syscall.Statfs_t{ + {Bsize: 0}, {Bsize: 4096, Blocks: 100, Bavail: 101}, {Bsize: 4096, Blocks: math.MaxUint64}, + } { + if _, _, _, err := ephemeralStorageSnapshot(invalid, 1); err == nil { + t.Fatalf("accepted invalid statfs: %+v", invalid) + } + } +} + +func TestEphemeralStorageSnapshotReadsConfiguredFilestore(t *testing.T) { + m := NewModule(t.TempDir(), "", false, 2) + capacity, available, disk, err := m.EphemeralStorageSnapshot() + if err != nil { + t.Fatal(err) + } + if capacity == 0 || available > capacity || disk == nil || *disk < 0 || *disk > 1 { + t.Fatalf("invalid snapshot: %d %d %v", capacity, available, disk) + } + m.FilestoreDir = filepath.Join(t.TempDir(), "missing") + if _, _, disk, err := m.EphemeralStorageSnapshot(); err == nil || disk != nil { + t.Fatalf("missing filestore: disk=%v err=%v", disk, err) + } +}