Add disk-load's cpu_top tool: CPU by systemd unit and container (issue 741)

No tool named which units or processes used CPU outside containers during
the novox overload (issue 739). cpu_top reads each cgroup's cpu.stat (no
root needed), ranks units or containers busiest first, and says the
machine's CPU in use and the busy time no unit accounts for.

modules/disk-load/cmd/disk-load/cgroups.go

@@ -68,9 +68,10 @@ func (m *Machine) bottomDevices() map[string]bool {
 	return out
 }
 
-// unitIO is every unit's storage I/O, by its cgroup's path: the innermost .service or .scope, because
-// io.stat counts a cgroup's descendants too and an outer unit would count them twice.
-func (m *Machine) unitIO(counted map[string]bool) map[string]cgroupIO {
+// unitPaths is every systemd unit's cgroup path under /sys/fs/cgroup: the innermost .service or
+// .scope, because a cgroup's own stat file counts its descendants too and an outer unit would count
+// them twice. Shared by every reader that walks the cgroup tree (unitIO here, unitCPU in cpu.go).
+func (m *Machine) unitPaths() []string {
 	base := m.path("/sys/fs/cgroup")
 	var units []string
 	filepath.WalkDir(base, func(p string, d fs.DirEntry, err error) error {
@@ -89,11 +90,19 @@ func (m *Machine) unitIO(counted map[string]bool) map[string]cgroupIO {
 			outer[dir] = true
 		}
 	}
-	out := map[string]cgroupIO{}
+	out := make([]string, 0, len(units))
 	for _, u := range units {
-		if outer[u] {
-			continue
+		if !outer[u] {
+			out = append(out, u)
 		}
+	}
+	return out
+}
+
+// unitIO is every unit's storage I/O, by its cgroup's path.
+func (m *Machine) unitIO(counted map[string]bool) map[string]cgroupIO {
+	out := map[string]cgroupIO{}
+	for _, u := range m.unitPaths() {
 		text := m.read("/sys/fs/cgroup/" + u + "/io.stat")
 		if text == "" {
 			continue

modules/disk-load/cmd/disk-load/cpu.go

@@ -0,0 +1,164 @@
+package main
+
+import (
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// cgroupCPU is a cgroup's processor time, from its cpu.stat.
+type cgroupCPU struct {
+	usageUsec uint64
+}
+
+// parseCPUStat reads a cgroup's cpu.stat ("usage_usec 123\nuser_usec …\nsystem_usec …"), readable by
+// every account, unlike /proc/<pid>/stat for another account's process.
+func parseCPUStat(s string) cgroupCPU {
+	var c cgroupCPU
+	for _, line := range strings.Split(s, "\n") {
+		k, v, ok := strings.Cut(line, " ")
+		if !ok {
+			continue
+		}
+		if k == "usage_usec" {
+			c.usageUsec, _ = strconv.ParseUint(strings.TrimSpace(v), 10, 64)
+		}
+	}
+	return c
+}
+
+// unitCPU is every unit's processor time, by its cgroup's path. A cgroup whose cpu.stat cannot be
+// read is left out, never an error for the whole answer.
+func (m *Machine) unitCPU() map[string]cgroupCPU {
+	out := map[string]cgroupCPU{}
+	for _, u := range m.unitPaths() {
+		text := m.read("/sys/fs/cgroup/" + u + "/cpu.stat")
+		if text == "" {
+			continue
+		}
+		out[u] = parseCPUStat(text)
+	}
+	return out
+}
+
+// MachineCPU is the machine's CPU in use over the window, from /proc/stat's cpu line.
+type MachineCPU struct {
+	CPUs     int     `json:"cpus"`
+	BusyPct  float64 `json:"busy_pct"`  // % of all CPUs together
+	CPUsBusy float64 `json:"cpus_busy"` // as a count of CPUs, e.g. 7.5 of 12
+}
+
+// machineCPUOver is the machine's CPU in use between two readings of /proc/stat: everything but idle
+// and iowait, counted against the window's total ticks over every CPU.
+func machineCPUOver(a, b cpuTimes) MachineCPU {
+	total := float64(b.total() - a.total())
+	if total <= 0 {
+		return MachineCPU{CPUs: b.cpus}
+	}
+	idle := sub(b.idle, a.idle) + sub(b.iowait, a.iowait)
+	busy := total - idle
+	if busy < 0 {
+		busy = 0
+	}
+	pct := round1(busy / total * 100)
+	return MachineCPU{CPUs: b.cpus, BusyPct: pct, CPUsBusy: round1(pct / 100 * float64(b.cpus))}
+}
+
+// CPUUsage is a unit's or a container's share of CPU between two readings of its cgroup's cpu.stat.
+type CPUUsage struct {
+	Unit      string  `json:"unit,omitempty"`
+	Container string  `json:"container,omitempty"`
+	CPUPct    float64 `json:"cpu_pct"`   // % of one CPU: 100 = one CPU busy the whole window
+	SharePct  float64 `json:"share_pct"` // this unit's share of the machine's busy time
+	raw       float64
+}
+
+// topCPU ranks the units, or the containers, with the most CPU time between two readings of
+// cpu.stat, busiest first. unaccountedPct is the machine's busy time no unit accounts for, as a
+// share of it (never below 0): the sum is always taken over every unit, whichever by ranks by, so a
+// limit on the list never changes it.
+func (m *Machine) topCPU(a, b map[string]cgroupCPU, window time.Duration, by string, limit int, cpusBusy float64) ([]CPUUsage, float64) {
+	windowUsec := window.Seconds() * 1e6
+	type entry struct {
+		unit, container string
+		pct             float64
+	}
+	var entries []entry
+	var names map[string]string
+	var sumPct float64
+	for path, end := range b {
+		start, ok := a[path]
+		if !ok {
+			continue
+		}
+		d := sub(end.usageUsec, start.usageUsec)
+		if d == 0 {
+			continue
+		}
+		pct := d / windowUsec * 100
+		sumPct += pct
+		unit := filepath.Base(path)
+		var container string
+		if id := containerID.FindStringSubmatch("/" + path); id != nil {
+			if names == nil {
+				names = m.containerNames()
+			}
+			container = id[1][:12]
+			if n, ok := names[id[1]]; ok {
+				container = n
+			}
+		}
+		entries = append(entries, entry{unit: unit, container: container, pct: pct})
+	}
+
+	groups := map[string]*CPUUsage{}
+	for _, e := range entries {
+		key := e.unit
+		if by == "container" {
+			if e.container == "" {
+				continue
+			}
+			key = e.container
+		}
+		g, ok := groups[key]
+		if !ok {
+			g = &CPUUsage{Unit: e.unit, Container: e.container}
+			if by == "container" {
+				g.Unit = ""
+			}
+			groups[key] = g
+		}
+		g.raw += e.pct
+	}
+	out := make([]CPUUsage, 0, len(groups))
+	for _, g := range groups {
+		out = append(out, *g)
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].raw != out[j].raw {
+			return out[i].raw > out[j].raw
+		}
+		return out[i].Unit+out[i].Container < out[j].Unit+out[j].Container
+	})
+	if len(out) > limit {
+		out = out[:limit]
+	}
+	for i := range out {
+		out[i].CPUPct = round1(out[i].raw)
+		if cpusBusy > 0 {
+			out[i].SharePct = round1(out[i].raw / (cpusBusy * 100) * 100)
+		}
+	}
+
+	unaccounted := cpusBusy*100 - sumPct
+	if unaccounted < 0 {
+		unaccounted = 0
+	}
+	unaccountedPct := 0.0
+	if cpusBusy > 0 {
+		unaccountedPct = round1(unaccounted / (cpusBusy * 100) * 100)
+	}
+	return out, unaccountedPct
+}

modules/disk-load/cmd/disk-load/cpu_test.go

@@ -0,0 +1,107 @@
+package main
+
+import (
+	"strings"
+	"testing"
+	"time"
+)
+
+// aCPUHog is a machine whose CPU is 75 % busy (1.5 of 2 CPUs) over a 2 s window: a unit, another unit,
+// and a container, each with a known usage_usec delta, leaving part of the machine's busy time
+// unaccounted — the shape of issue 739, where 750 % of 12 CPUs went to processes outside containers.
+func aCPUHog(t *testing.T) *fake {
+	f := newFake(t)
+	f.file("/proc/stat", "cpu  0 0 0 0 0 0 0 0 0 0\ncpu0 0 0 0 0 0 0 0 0 0 0\ncpu1 0 0 0 0 0 0 0 0 0 0\nintr 1\n")
+	f.then("/proc/stat", "cpu  100 0 50 50 0 0 0 0 0 0\ncpu0 50 0 25 25 0 0 0 0 0 0\ncpu1 50 0 25 25 0 0 0 0 0 0\nintr 2\n")
+
+	zero := "usage_usec 0\nuser_usec 0\nsystem_usec 0\n"
+	cg := func(path string, usec int) {
+		f.file("/sys/fs/cgroup/"+path+"/cpu.stat", zero)
+		f.then("/sys/fs/cgroup/"+path+"/cpu.stat", "usage_usec "+itoa(usec)+"\nuser_usec "+itoa(usec)+"\nsystem_usec 0\n")
+	}
+	cg("system.slice/alpha.service", 1000000)         // 50 % of one CPU over 2 s
+	cg("system.slice/beta.service", 400000)           // 20 %
+	cg("system.slice/docker-"+store+".scope", 600000) // 30 %, named "cpu-hog" by the runtime
+	f.answers["docker ps --no-trunc --format {{.ID}}\t{{.Names}}"] = store + "\tcpu-hog\n"
+	return f
+}
+
+func TestCPUTopRanksUnitsAndContainersByCPUOverTheWindow(t *testing.T) {
+	f := aCPUHog(t)
+	out, err := tool(t, f.machine(), "cpu_top").Run(map[string]any{"seconds": 2.0})
+	if err != nil {
+		t.Fatal(err)
+	}
+	m := out.(map[string]any)
+	if m["by"] != "unit" || f.waited != 2*time.Second {
+		t.Errorf("the default: %+v, waited %s", m, f.waited)
+	}
+	cpu := m["cpu"].(MachineCPU)
+	if cpu.CPUs != 2 || cpu.BusyPct != 75 || cpu.CPUsBusy != 1.5 {
+		t.Errorf("machine cpu %+v", cpu)
+	}
+	top := m["top"].([]CPUUsage)
+	if len(top) != 3 {
+		t.Fatalf("three units moved: %+v", top)
+	}
+	if top[0].Unit != "alpha.service" || top[0].CPUPct != 50 || top[0].SharePct != 33.3 {
+		t.Errorf("busiest first: %+v", top[0])
+	}
+	if top[1].Container != "cpu-hog" || top[1].CPUPct != 30 || top[1].SharePct != 20 {
+		t.Errorf("the container, named by the runtime: %+v", top[1])
+	}
+	if top[2].Unit != "beta.service" || top[2].CPUPct != 20 || top[2].SharePct != 13.3 {
+		t.Errorf("the least busy: %+v", top[2])
+	}
+	if m["unaccounted_pct"] != 33.3 {
+		t.Errorf("unaccounted: %+v (machine busy 150%% of one CPU, units sum to 100%%, 50%% of 150%% left over)", m["unaccounted_pct"])
+	}
+}
+
+func TestCPUTopByContainerNamesOnlyContainers(t *testing.T) {
+	out, err := tool(t, aCPUHog(t).machine(), "cpu_top").Run(map[string]any{"by": "container", "seconds": 2.0})
+	if err != nil {
+		t.Fatal(err)
+	}
+	top := out.(map[string]any)["top"].([]CPUUsage)
+	if len(top) != 1 || top[0].Container != "cpu-hog" || top[0].Unit != "" || top[0].CPUPct != 30 {
+		t.Errorf("by container: %+v", top)
+	}
+	if _, err := tool(t, aCPUHog(t).machine(), "cpu_top").Run(map[string]any{"by": "process"}); err == nil {
+		t.Error("by=process was accepted; cpu_top only ranks units and containers")
+	}
+}
+
+func TestCPUTopLimitsHowManyAreNamed(t *testing.T) {
+	out, err := tool(t, aCPUHog(t).machine(), "cpu_top").Run(map[string]any{"limit": 1.0, "seconds": 2.0})
+	if err != nil {
+		t.Fatal(err)
+	}
+	m := out.(map[string]any)
+	top := m["top"].([]CPUUsage)
+	if len(top) != 1 || top[0].Unit != "alpha.service" {
+		t.Errorf("limit 1: %+v", top)
+	}
+	// The limit narrows the list, never the unaccounted share, which is taken over every unit.
+	if m["unaccounted_pct"] != 33.3 {
+		t.Errorf("unaccounted with limit 1: %+v", m["unaccounted_pct"])
+	}
+}
+
+func TestCPUTopLeavesOutACgroupThatAppearsOrVanishesWithinTheWindow(t *testing.T) {
+	f := aCPUHog(t)
+	f.then("/sys/fs/cgroup/system.slice/short-lived.service/cpu.stat", "usage_usec 500000\n")
+	out, err := tool(t, f.machine(), "cpu_top").Run(map[string]any{"seconds": 2.0})
+	if err != nil {
+		t.Fatal(err)
+	}
+	top := out.(map[string]any)["top"].([]CPUUsage)
+	for _, u := range top {
+		if u.Unit == "short-lived.service" {
+			t.Errorf("a unit only in the second reading was counted: %+v", top)
+		}
+	}
+	if !strings.Contains(strings.Join(f.calls, ","), "docker ps") {
+		t.Errorf("the container runtime was never asked: %v", f.calls)
+	}
+}

modules/disk-load/cmd/disk-load/tools.go

@@ -85,6 +85,57 @@ func Tools(m *Machine) []stdio.Tool {
 				return out, nil
 			},
 		},
+		{
+			Name: "cpu_top",
+			Description: "What used the most CPU over a few seconds, from each unit's cgroup cpu.stat, which needs no root: the machine's " +
+				"CPU in use (as % of all CPUs and as CPUs busy, from /proc/stat), then by=unit (the default) each systemd unit and " +
+				"container ranked by CPU, busiest first, with the busy time no unit accounts for; by=container only the containers, " +
+				"by name. Replaces top, systemd-cgtop. (r)",
+			Input: map[string]any{
+				"seconds": map[string]any{"type": "integer", "description": "how long to sample, in seconds (default 3, at most 10)"},
+				"limit":   map[string]any{"type": "integer", "description": "how many units or containers to name (default 15, at most 50)"},
+				"by":      map[string]any{"type": "string", "enum": []string{"unit", "container"}, "description": "what to rank (default unit)"},
+			},
+			Run: func(args map[string]any) (any, error) {
+				n, err := count(args, "seconds", 3, 1, 10)
+				if err != nil {
+					return nil, err
+				}
+				w := time.Duration(n) * time.Second
+				limit, err := count(args, "limit", 15, 1, 50)
+				if err != nil {
+					return nil, err
+				}
+				by := text(args, "by")
+				switch by {
+				case "":
+					by = "unit"
+				case "unit", "container":
+				default:
+					return nil, fmt.Errorf("by is unit or container, not %q", by)
+				}
+				ac, err := parseCPU(m.read("/proc/stat"))
+				if err != nil {
+					return nil, err
+				}
+				a := m.unitCPU()
+				m.Wait(w)
+				bc, err := parseCPU(m.read("/proc/stat"))
+				if err != nil {
+					return nil, err
+				}
+				b := m.unitCPU()
+				machine := machineCPUOver(ac, bc)
+				top, unaccounted := m.topCPU(a, b, w, by, limit, machine.CPUsBusy)
+				return map[string]any{
+					"window_s":        w.Seconds(),
+					"by":              by,
+					"cpu":             machine,
+					"top":             top,
+					"unaccounted_pct": unaccounted,
+				}, nil
+			},
+		},
 		{
 			Name: "disk_filesystems",
 			Description: "The space of every filesystem that stores data (not proc, tmpfs or overlay layers): size, used and free in GiB, " +