diff --git a/runner/internal/runner/metrics/cgroups.go b/runner/internal/runner/metrics/cgroups.go index 7ac89db4a..8769d8a58 100644 --- a/runner/internal/runner/metrics/cgroups.go +++ b/runner/internal/runner/metrics/cgroups.go @@ -6,11 +6,29 @@ import ( "errors" "fmt" "os" + "path" "strings" "github.com/dstackai/dstack/runner/internal/common/log" ) +func getMetricsCgroupPath(ctx context.Context, mountPoint, procPidCgroupPath string) (string, error) { + // In a container cgroup namespace, the mount root accounts for the whole + // container, including nested containers in sibling groups of /dind. + // The host's root cgroup has no memory.current; keep using the process + // cgroup in that case. + if _, err := os.Stat(path.Join(mountPoint, "memory.current")); err == nil { + return mountPoint, nil + } else if !errors.Is(err, os.ErrNotExist) { + return "", fmt.Errorf("stat cgroup root memory.current: %w", err) + } + cgroupPathname, err := getProcessCgroupPathname(ctx, procPidCgroupPath) + if err != nil { + return "", fmt.Errorf("get cgroup pathname: %w", err) + } + return path.Join(mountPoint, cgroupPathname), nil +} + func getProcessCgroupMountPoint(ctx context.Context, ProcPidMountsPath string) (string, error) { // See proc_pid_mounts(5) for the ProcPidMountsPath file description file, err := os.Open(ProcPidMountsPath) diff --git a/runner/internal/runner/metrics/cgroups_test.go b/runner/internal/runner/metrics/cgroups_test.go index 3e6e0abca..a86ac8589 100644 --- a/runner/internal/runner/metrics/cgroups_test.go +++ b/runner/internal/runner/metrics/cgroups_test.go @@ -15,6 +15,51 @@ const ( rootMountLine = "/dev/nvme0n1p5 / ext4 rw,relatime 0 0" ) +func TestGetMetricsCgroupPath(t *testing.T) { + for _, tc := range []struct { + name string + processGroup string + rootMemory bool + childMemory bool + wantGroup string + }{ + {name: "ordinary container", processGroup: "/", rootMemory: true, wantGroup: "/"}, + {name: "dind before nested container", processGroup: "/dind", rootMemory: true, wantGroup: "/"}, + {name: "dind with nested container", processGroup: "/dind", rootMemory: true, childMemory: true, wantGroup: "/"}, + {name: "host cgroup namespace", processGroup: "/system.slice/docker.scope", childMemory: true, wantGroup: "/system.slice/docker.scope"}, + } { + t.Run(tc.name, func(t *testing.T) { + mountPoint := t.TempDir() + procFile := createProcFile(t, "cgroup", "0::"+tc.processGroup) + if tc.rootMemory { + require.NoError(t, os.WriteFile(path.Join(mountPoint, "memory.current"), []byte("8192\n"), 0o600)) + } + if tc.childMemory { + child := path.Join(mountPoint, tc.processGroup) + require.NoError(t, os.MkdirAll(child, 0o700)) + require.NoError(t, os.WriteFile(path.Join(child, "memory.current"), []byte("1024\n"), 0o600)) + } + cgroupPath, err := getMetricsCgroupPath(t.Context(), mountPoint, procFile) + require.NoError(t, err) + require.Equal(t, path.Join(mountPoint, tc.wantGroup), cgroupPath) + collector := &MetricsCollector{} + memory, err := collector.GetMemoryUsageBytes(cgroupPath) + require.NoError(t, err) + if tc.rootMemory { + require.Equal(t, uint64(8192), memory) + } else { + require.Equal(t, uint64(1024), memory) + } + }) + } +} + +func TestGetMetricsCgroupPath_ErrorMissingProcessCgroup(t *testing.T) { + _, err := getMetricsCgroupPath(t.Context(), t.TempDir(), path.Join(t.TempDir(), "missing")) + require.ErrorContains(t, err, "get cgroup pathname") + require.ErrorIs(t, err, os.ErrNotExist) +} + func TestGetProcessCgroupMountPoint_ErrorNoCgroupMounts(t *testing.T) { procPidMountsPath := createProcFile(t, "mounts", rootMountLine, "malformed line") diff --git a/runner/internal/runner/metrics/metrics.go b/runner/internal/runner/metrics/metrics.go index 34f22c731..aee9c0b74 100644 --- a/runner/internal/runner/metrics/metrics.go +++ b/runner/internal/runner/metrics/metrics.go @@ -37,13 +37,12 @@ func NewMetricsCollector(ctx context.Context) (*MetricsCollector, error) { } func (s *MetricsCollector) GetSystemMetrics(ctx context.Context) (*schemas.SystemMetrics, error) { - // It's possible to move a process from one control group to another (it's unlikely, but nonetheless), - // so we detect the current group each time. - cgroupPathname, err := getProcessCgroupPathname(ctx, "/proc/self/cgroup") + // Resolve the accounting group each time: start-dockerd can move the runner + // into a child cgroup, and host-namespace processes can change groups too. + cgroupPath, err := getMetricsCgroupPath(ctx, s.cgroupMountPoint, "/proc/self/cgroup") if err != nil { - return nil, fmt.Errorf("get cgroup pathname: %w", err) + return nil, err } - cgroupPath := path.Join(s.cgroupMountPoint, cgroupPathname) timestamp := time.Now() cpuUsage, err := s.GetCPUUsageMicroseconds(cgroupPath) if err != nil { diff --git a/runner/internal/runner/metrics/metrics_test.go b/runner/internal/runner/metrics/metrics_test.go index 844b02bd3..d9b076519 100644 --- a/runner/internal/runner/metrics/metrics_test.go +++ b/runner/internal/runner/metrics/metrics_test.go @@ -1,12 +1,35 @@ package metrics import ( + "os" + "path" "testing" - "github.com/dstackai/dstack/runner/internal/runner/schemas" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/dstackai/dstack/runner/internal/common/gpu" + "github.com/dstackai/dstack/runner/internal/runner/schemas" ) +func TestGetSystemMetrics_ContainerRoot(t *testing.T) { + mountPoint := t.TempDir() + for name, content := range map[string]string{ + "cpu.stat": "usage_usec 12345\nuser_usec 12000\nsystem_usec 345\n", + "memory.current": "8192\n", + "memory.stat": "anon 6144\ninactive_file 2048\n", + } { + require.NoError(t, os.WriteFile(path.Join(mountPoint, name), []byte(content), 0o600)) + } + collector := &MetricsCollector{cgroupMountPoint: mountPoint, gpuVendor: gpu.GpuVendorNone} + metrics, err := collector.GetSystemMetrics(t.Context()) + require.NoError(t, err) + require.Equal(t, uint64(12345), metrics.CpuUsage) + require.Equal(t, uint64(8192), metrics.MemoryUsage) + require.Equal(t, uint64(6144), metrics.MemoryWorkingSet) + require.Empty(t, metrics.GPUMetrics) +} + func TestGetAMDGPUMetrics_OK(t *testing.T) { collector, err := NewMetricsCollector(t.Context()) assert.NoError(t, err)