#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Inspect and modify Linux cgroup v2 hierarchies.""" from __future__ import annotations import argparse import locale import os import re import stat import sys import time from collections.abc import Sequence from dataclasses import dataclass, field from decimal import Decimal, InvalidOperation from enum import StrEnum from functools import cmp_to_key from pathlib import Path from typing import Any ####### Program constants ######## PROGRAM_VERSION = "0.0.1a" TREE_VERTICAL = "\u2502" # │ TREE_HORIZONTAL = "\u2500" # ─ TREE_DOWN = "\u252c" # ┬ TREE_BRANCH = "\u251c" # ├ TREE_LAST = "\u2514" # └ ########################################################################### # Core types and data structures ########################################################################### # These classes define the shared types and data structures used throughout # cgtree. class CgtreeError(Exception): """Error reported to the user without a traceback.""" class InterfaceFormat(StrEnum): """Supported cgroup interface formats.""" SCALAR = "scalar" FLAT = "flat-keyed" NESTED = "nested-keyed" LIST = "list" class UnitKind(StrEnum): """Supported display units for parsed cgroup values.""" BYTES = "bytes" PAGES = "pages" USEC = "usec" PERCENT = "percent" COUNT = "count" OPS = "ops" STRING = "string" class Align(StrEnum): """Supported table-cell alignments.""" LEFT = "left" RIGHT = "right" ####### Data classes ######## @dataclass class MountInfo: """Detected cgroup v2 mount point and superblock options.""" mount_point: Path options: tuple[str, ...] @dataclass class CgroupNode: """One directory in the cgroup tree.""" path_abs: Path path_rel: str name: str parent: CgroupNode | None = None children: list[CgroupNode] = field(default_factory=list) @dataclass(frozen=True) class FileSpec: """Metadata describing one cgroup interface file.""" name: str interface_format: InterfaceFormat default_unit: UnitKind = UnitKind.STRING field_units: dict[str, UnitKind] = field(default_factory=dict) device_keyed: bool = False writable: bool = False @dataclass(frozen=True) class InterfaceMetadata: """Metadata describing one flat- or nested-keyed cgroup interface.""" default_unit: UnitKind = UnitKind.COUNT field_units: dict[str, UnitKind] = field(default_factory=dict) display_fields: tuple[str, ...] = () device_keyed: bool = False discover_display_fields: bool = False @dataclass(frozen=True) class SelectorSpec: """Resolved metadata for one display selector.""" name: str header: str file_name: str | None = None key_name: str | None = None field_name: str | None = None unit: UnitKind = UnitKind.STRING full_file: bool = False aggregate: bool = False virtual: bool = False header_lines: tuple[str, ...] = () @dataclass class DisplayCell: """Formatted table cell and its underlying sort value.""" text: str = "" sort_value: int | float | str | None = None align: Align = Align.RIGHT @dataclass class DisplayRow: """Data for one table row, including its tree label and optional key.""" label: str cells: dict[str, DisplayCell] key: str = "" dev: str = "" is_context: bool = False @dataclass(frozen=True) class ProcessInfo: """Process ID and command name read from /proc/PID/comm.""" pid: int comm: str = "" @dataclass class DisplayOptions: """Options controlling row construction and display.""" compact: bool dev_format: str pretty_headers: bool @dataclass class Assignment: """Parsed --set assignment and optional target cgroup.""" file_name: str raw_value: str target_path_rel: str | None = None ########################################################################### # Interface metadata ########################################################################### # Supported interfaces are added in format-specific groups. # # HugeTLB interfaces have architecture-dependent sizes, so all # hugetlb..* files are classified separately by HUGETLB_DYNAMIC_RE. ####### Scalar interfaces ######## # Scalar interfaces are read as one value or raw line. BYTE_SCALAR_FILES = { "memory.current", "memory.high", "memory.low", "memory.max", "memory.min", "memory.peak", "memory.swap.current", "memory.swap.high", "memory.swap.max", "memory.swap.peak", "memory.zswap.current", "memory.zswap.max", } COUNT_SCALAR_FILES = { "cgroup.freeze", "cgroup.kill", "cgroup.max.depth", "cgroup.max.descendants", "cgroup.pressure", "cpu.idle", "cpu.max.burst", "cpu.weight", "cpu.weight.nice", "memory.oom.group", "memory.zswap.writeback", "pids.current", "pids.max", "pids.peak", } PERCENT_SCALAR_FILES = { "cpu.uclamp.min", "cpu.uclamp.max", } # These scalar interfaces contain structured or textual values that are # displayed without unit formatting. STRING_SCALAR_FILES = { "cgroup.type", "cpu.max", "cpuset.cpus", "cpuset.cpus.effective", "cpuset.cpus.exclusive", "cpuset.cpus.exclusive.effective", "cpuset.cpus.isolated", "cpuset.cpus.partition", "cpuset.mems", "cpuset.mems.effective", "io.prio.class", } ####### List interfaces ######## # Whitespace-separated values. CONTROLLER_LIST_FILES = { "cgroup.controllers", "cgroup.subtree_control", } # Line-separated values. ID_LIST_FILES = { "cgroup.procs", "cgroup.threads", } LIST_FILES = CONTROLLER_LIST_FILES | ID_LIST_FILES ####### Flat-keyed interfaces ######## # memory.stat contains mixed units. Kernel documentation defines memory-sized # fields in bytes and reclaim or activity fields as page counts. Unlisted # fields are displayed as strings. MEMORY_STAT_BYTE_FIELDS = { "anon", "file", "kernel", "kernel_stack", "pagetables", "sec_pagetables", "percpu", "sock", "vmalloc", "shmem", "zswap", "zswapped", "file_mapped", "file_dirty", "file_writeback", "swapcached", "anon_thp", "file_thp", "shmem_thp", "inactive_anon", "active_anon", "inactive_file", "active_file", "unevictable", "slab", "slab_reclaimable", "slab_unreclaimable", "hugetlb", } MEMORY_STAT_PAGE_FIELDS = { "pswpin", "pswpout", "pgscan", "pgscan_direct", "pgscan_khugepaged", "pgscan_kswapd", "pgscan_proactive", "pgsteal", "pgsteal_direct", "pgsteal_khugepaged", "pgsteal_kswapd", "pgsteal_proactive", "pgrefill", "pgactivate", "pgdeactivate", "pglazyfree", "pglazyfreed", "swpin_zero", "swpout_zero", "zswpin", "zswpout", "zswpwb", "numa_pages_migrated", "pgdemote_direct", "pgdemote_khugepaged", "pgdemote_kswapd", "pgdemote_proactive", } # Flat-keyed interfaces contain one KEY VALUE pair per line. # # Every entry names a supported interface. Optional metadata describes field # units or block-device keys when additional handling is required. FLAT_INTERFACES: dict[str, InterfaceMetadata] = { "cgroup.events": InterfaceMetadata(), "cgroup.stat": InterfaceMetadata(), "cgroup.stat.local": InterfaceMetadata(), "cpu.stat": InterfaceMetadata( field_units={ "usage_usec": UnitKind.USEC, "user_usec": UnitKind.USEC, "system_usec": UnitKind.USEC, }, ), "cpu.stat.local": InterfaceMetadata( field_units={"throttled_usec": UnitKind.USEC}, ), "io.bfq.weight": InterfaceMetadata(device_keyed=True), "io.weight": InterfaceMetadata(device_keyed=True), "memory.events": InterfaceMetadata(), "memory.events.local": InterfaceMetadata(), "memory.stat": InterfaceMetadata( field_units={ **{name: UnitKind.BYTES for name in MEMORY_STAT_BYTE_FIELDS}, **{name: UnitKind.PAGES for name in MEMORY_STAT_PAGE_FIELDS}, }, ), "memory.swap.events": InterfaceMetadata(), "misc.capacity": InterfaceMetadata(), "misc.current": InterfaceMetadata(), "misc.events": InterfaceMetadata(), "misc.events.local": InterfaceMetadata(), "misc.max": InterfaceMetadata(), "misc.peak": InterfaceMetadata(), "pids.events": InterfaceMetadata(), "pids.events.local": InterfaceMetadata(), } ####### Nested-keyed interfaces ######## # Pressure stall information (PSI) interfaces use "some" and "full" rows with # the same fields. PRESSURE_FIELDS = ("avg10", "avg60", "avg300", "total") PRESSURE_FIELD_UNITS = { "avg10": UnitKind.PERCENT, "avg60": UnitKind.PERCENT, "avg300": UnitKind.PERCENT, "total": UnitKind.USEC, } RDMA_EVENT_FIELDS = ( "hca_handle.max", "hca_handle.alloc_fail", "hca_object.max", "hca_object.alloc_fail", ) RDMA_EVENT_FIELD_UNITS = {name: UnitKind.COUNT for name in RDMA_EVENT_FIELDS} # Nested-keyed interfaces contain one row key followed by FIELD=VALUE pairs. # # Every entry names a supported interface. Its metadata describes field units, # whole-interface display fields, block-device keys, and any runtime field # discovery. DMEM is a special case that uses one keyed scalar value instead # of FIELD=VALUE pairs. NESTED_INTERFACES: dict[str, InterfaceMetadata] = { "cpu.pressure": InterfaceMetadata( field_units=PRESSURE_FIELD_UNITS, display_fields=PRESSURE_FIELDS, ), "dmem.capacity": InterfaceMetadata( field_units={"value": UnitKind.BYTES}, display_fields=("value",), ), "dmem.current": InterfaceMetadata( field_units={"value": UnitKind.BYTES}, display_fields=("value",), ), "dmem.low": InterfaceMetadata( field_units={"value": UnitKind.BYTES}, display_fields=("value",), ), "dmem.max": InterfaceMetadata( field_units={"value": UnitKind.BYTES}, display_fields=("value",), ), "dmem.min": InterfaceMetadata( field_units={"value": UnitKind.BYTES}, display_fields=("value",), ), "io.cost.model": InterfaceMetadata( field_units={ "rbps": UnitKind.BYTES, "wbps": UnitKind.BYTES, "rseqiops": UnitKind.OPS, "wseqiops": UnitKind.OPS, "rrandiops": UnitKind.OPS, "wrandiops": UnitKind.OPS, }, display_fields=( "ctrl", "model", "rbps", "wbps", "rseqiops", "wseqiops", "rrandiops", "wrandiops", ), device_keyed=True, ), "io.cost.qos": InterfaceMetadata( field_units={ "enable": UnitKind.COUNT, "rpct": UnitKind.PERCENT, "rlat": UnitKind.USEC, "wpct": UnitKind.PERCENT, "wlat": UnitKind.USEC, "min": UnitKind.PERCENT, "max": UnitKind.PERCENT, }, display_fields=("enable", "ctrl", "rpct", "rlat", "wpct", "wlat", "min", "max"), device_keyed=True, ), "io.latency": InterfaceMetadata( display_fields=("target",), device_keyed=True, ), "io.max": InterfaceMetadata( field_units={ "rbps": UnitKind.BYTES, "wbps": UnitKind.BYTES, "riops": UnitKind.OPS, "wiops": UnitKind.OPS, }, display_fields=("rbps", "wbps", "riops", "wiops"), device_keyed=True, ), "io.pressure": InterfaceMetadata( field_units=PRESSURE_FIELD_UNITS, display_fields=PRESSURE_FIELDS, ), "io.stat": InterfaceMetadata( field_units={ "rbytes": UnitKind.BYTES, "wbytes": UnitKind.BYTES, "dbytes": UnitKind.BYTES, "rios": UnitKind.OPS, "wios": UnitKind.OPS, "dios": UnitKind.OPS, }, display_fields=("rbytes", "wbytes", "rios", "wios", "dbytes", "dios"), device_keyed=True, ), "irq.pressure": InterfaceMetadata( field_units=PRESSURE_FIELD_UNITS, display_fields=PRESSURE_FIELDS, ), "memory.numa_stat": InterfaceMetadata( default_unit=UnitKind.BYTES, # Placeholder used until runtime discovery determines the available # NUMA node fields. display_fields=("N0",), discover_display_fields=True, ), "memory.pressure": InterfaceMetadata( field_units=PRESSURE_FIELD_UNITS, display_fields=PRESSURE_FIELDS, ), "memory.reclaim": InterfaceMetadata( field_units={"swappiness": UnitKind.COUNT}, display_fields=("swappiness",), ), "rdma.current": InterfaceMetadata( display_fields=("hca_handle", "hca_object"), ), "rdma.events": InterfaceMetadata( field_units=RDMA_EVENT_FIELD_UNITS, display_fields=RDMA_EVENT_FIELDS, ), "rdma.events.local": InterfaceMetadata( field_units=RDMA_EVENT_FIELD_UNITS, display_fields=RDMA_EVENT_FIELDS, ), "rdma.max": InterfaceMetadata( display_fields=("hca_handle", "hca_object"), ), "rdma.peak": InterfaceMetadata( field_units={ "hca_handle": UnitKind.COUNT, "hca_object": UnitKind.COUNT, }, display_fields=("hca_handle", "hca_object"), ), } ####### Dynamic interfaces ######## # Some interface names and fields depend on the running kernel configuration. # # HugeTLB creates one hugetlb..* family for each supported huge-page # size. NUMA-aware interfaces expose node fields named N0, N1, N2, and so on. HUGETLB_DYNAMIC_RE = re.compile( r"^hugetlb\.(?P[A-Za-z0-9]+)\." r"(?Pcurrent|max|events|events\.local|numa_stat|rsvd\.current|rsvd\.max)$" ) NUMA_NODE_FIELD_RE = re.compile(r"^N(?P[0-9]+)$") HUGETLB_LIMIT_WRITE_SCHEMA = {"kind": "single", "value": "bytes_or_max"} def hugetlb_interface_kind(file_name: str) -> str | None: """Return the kind suffix for one hugetlb..* interface.""" match = HUGETLB_DYNAMIC_RE.fullmatch(file_name) return match.group("kind") if match else None def nested_display_fields( file_name: str, discovered_fields: dict[str, tuple[str, ...]] | None = None, ) -> tuple[str, ...]: """Return display fields for a nested interface, preferring discovered fields.""" if discovered_fields is not None and file_name in discovered_fields: return discovered_fields[file_name] metadata = NESTED_INTERFACES.get(file_name) return metadata.display_fields if metadata is not None else () ####### Writable interfaces ######## # These schemas list the cgroup interfaces that cgtree can modify. # # Each schema defines the accepted --set syntax and how input is validated and # normalised before it is written to the kernel. WRITABLE_SCHEMAS: dict[str, dict[str, Any]] = { "cgroup.type": {"kind": "single", "value": "cgroup_type"}, "cpu.weight": {"kind": "single", "value": "int", "min": 1, "max": 10000}, "cpu.weight.nice": {"kind": "single", "value": "int", "min": -20, "max": 19}, "cpu.idle": {"kind": "single", "value": "int", "min": 0, "max": 1}, "cpu.max": {"kind": "cpu_max"}, "cpu.max.burst": {"kind": "single", "value": "int", "min": 0}, "cpu.uclamp.min": {"kind": "single", "value": "uclamp_percent"}, "cpu.uclamp.max": {"kind": "single", "value": "uclamp_percent_or_max"}, "cpuset.cpus": {"kind": "single", "value": "raw"}, "cpuset.mems": {"kind": "single", "value": "raw"}, "cpuset.cpus.partition": {"kind": "single", "value": "cpuset_partition"}, "io.cost.qos": { "kind": "nested_keyed", "fields": { "enable": "int", "ctrl": "raw", "rpct": "uclamp_percent", "rlat": "int", "wpct": "uclamp_percent", "wlat": "int", "min": "io_cost_percent", "max": "io_cost_percent", }, }, "io.cost.model": { "kind": "nested_keyed", "fields": { "ctrl": "raw", "model": "raw", "rbps": "bytes", "wbps": "bytes", "rseqiops": "int", "wseqiops": "int", "rrandiops": "int", "wrandiops": "int", }, }, "io.prio.class": {"kind": "single", "value": "raw"}, "memory.high": {"kind": "single", "value": "bytes_or_max"}, "memory.low": {"kind": "single", "value": "bytes_or_max"}, "memory.max": {"kind": "single", "value": "bytes_or_max"}, "memory.min": {"kind": "single", "value": "bytes_or_max"}, "memory.peak": {"kind": "single", "value": "raw"}, "memory.reclaim": {"kind": "memory_reclaim"}, "memory.swap.high": {"kind": "single", "value": "bytes_or_max"}, "memory.swap.max": {"kind": "single", "value": "bytes_or_max"}, "memory.swap.peak": {"kind": "single", "value": "raw"}, "memory.zswap.max": {"kind": "single", "value": "bytes_or_max"}, "memory.zswap.writeback": {"kind": "single", "value": "int", "min": 0, "max": 1}, "memory.oom.group": {"kind": "single", "value": "int", "min": 0, "max": 1}, "pids.max": {"kind": "single", "value": "int_or_max"}, "cgroup.freeze": {"kind": "single", "value": "int", "min": 0, "max": 1}, "cgroup.kill": {"kind": "single", "value": "int", "min": 1, "max": 1}, "cgroup.max.depth": {"kind": "single", "value": "int_or_max"}, "cgroup.max.descendants": {"kind": "single", "value": "int_or_max"}, "cgroup.pressure": {"kind": "single", "value": "int", "min": 0, "max": 1}, "cgroup.subtree_control": {"kind": "single", "value": "raw"}, "dmem.max": {"kind": "dmem_limit", "value": "bytes_or_max"}, "dmem.min": {"kind": "dmem_limit", "value": "bytes_or_max"}, "dmem.low": {"kind": "dmem_limit", "value": "bytes_or_max"}, "io.weight": {"kind": "flat_keyed", "value": "int_or_default", "min": 1, "max": 10000}, "io.bfq.weight": {"kind": "flat_keyed", "value": "int_or_default", "min": 1, "max": 10000}, "io.max": { "kind": "nested_keyed", "fields": { "rbps": "bytes_or_max", "wbps": "bytes_or_max", "riops": "int_or_max", "wiops": "int_or_max", }, }, "io.latency": {"kind": "nested_keyed", "fields": {"target": "int"}}, "rdma.max": { "kind": "nested_keyed", "fields": { "hca_handle": "int_or_max", "hca_object": "int_or_max", }, }, "misc.max": { "kind": "flat_keyed", "value": "int_or_max", "device_keyed": False, }, } ####### Runtime interface lookup ######## # FILE_SPECS combines the declarations above into a uniform representation # used for runtime lookups. def build_file_specs() -> dict[str, FileSpec]: """Build runtime interface specifications from the metadata tables.""" specs: dict[str, FileSpec] = {} # Combine the format-specific declarations into a uniform FileSpec # representation. for names, unit in ( (BYTE_SCALAR_FILES, UnitKind.BYTES), (COUNT_SCALAR_FILES, UnitKind.COUNT), (PERCENT_SCALAR_FILES, UnitKind.PERCENT), (STRING_SCALAR_FILES, UnitKind.STRING), ): for name in names: specs[name] = FileSpec( name=name, interface_format=InterfaceFormat.SCALAR, default_unit=unit, ) for name in LIST_FILES: specs[name] = FileSpec( name=name, interface_format=InterfaceFormat.LIST, ) for name, metadata in FLAT_INTERFACES.items(): specs[name] = FileSpec( name=name, interface_format=InterfaceFormat.FLAT, default_unit=metadata.default_unit, field_units=metadata.field_units, device_keyed=metadata.device_keyed, ) for name, metadata in NESTED_INTERFACES.items(): specs[name] = FileSpec( name=name, interface_format=InterfaceFormat.NESTED, default_unit=metadata.default_unit, field_units=metadata.field_units, device_keyed=metadata.device_keyed, ) # Read and write metadata are maintained separately because they describe # different aspects of an interface. Every writable interface must also be # listed above. missing_interfaces = set(WRITABLE_SCHEMAS) - set(specs) if missing_interfaces: names = ", ".join(sorted(missing_interfaces)) raise RuntimeError(f"writable interfaces are missing from the metadata tables: {names}") # Mark interfaces with a supported --set schema as writable. for name in WRITABLE_SCHEMAS: spec = specs[name] specs[name] = FileSpec( name=spec.name, interface_format=spec.interface_format, default_unit=spec.default_unit, field_units=spec.field_units, device_keyed=spec.device_keyed, writable=True, ) return specs FILE_SPECS = build_file_specs() ####### Built-in display headers ######## # Built-in views use friendly column headings. Selectors supplied through -C # retain their interface-based names. PRETTY_HEADERS = { "memory.current": "MEM", "memory.peak": "MEM PEAK", "memory.swap.current": "SWAP", "memory.swap.peak": "SWAP PEAK", "memory.max": "MEM MAX", "memory.high": "MEM HIGH", "io.stat.rbytes": "IO READ", "io.stat.wbytes": "IO WRITE", "io.stat.rios": "READ OPS", "io.stat.wios": "WRITE OPS", "io.weight": "IO WEIGHT", "io.weight.default": "IO WEIGHT", "cpu.percent": "CPU %", "cpu.stat.usage_usec": "CPU TIME", "cpu.stat.user_usec": "USER TIME", "cpu.stat.system_usec": "SYS TIME", "cpu.max": "CPU MAX", "cpu.max.percent": "CPU MAX", "cpu.weight": "WEIGHT", "cpu.weight.nice": "NICE", "cgroup.procs.count": "PROCS", "pids.current": "TASKS", "pids.max": "TASKS MAX", "pids.peak": "TASKS PEAK", } # Map display headings back to selectors so options such as --sort MEM can # resolve memory.current. HEADER_ALIASES = {value: key for key, value in PRETTY_HEADERS.items()} HEADER_ALIASES.update( { "CGROUP": "CGROUP", "KEY": "KEY", "DEV": "DEV", "CPU TIME": "cpu.stat.usage_usec", "USER TIME": "cpu.stat.user_usec", "SYS TIME": "cpu.stat.system_usec", } ) ####### Built-in views ######## DEFAULT_VIEW_SELECTORS = ( "memory.current", "memory.swap.current", "io.stat.rbytes", "io.stat.wbytes", "cpu.stat.usage_usec", "cpu.weight.nice", ) MEMORY_VIEW_SELECTORS = ( "memory.current", "memory.peak", "memory.swap.current", "memory.swap.peak", "memory.max", "memory.high", ) CPU_VIEW_SELECTORS = ( "cpu.percent", "cpu.stat.usage_usec", "cpu.stat.user_usec", "cpu.stat.system_usec", "cpu.weight", "cpu.weight.nice", "cpu.max.percent", ) IO_VIEW_SELECTORS = ( "io.stat.rbytes", "io.stat.wbytes", "io.stat.rios", "io.stat.wios", "io.weight.default", ) PIDS_VIEW_SELECTORS = ("cgroup.procs.count", "pids.current", "pids.peak", "pids.max") ########################################################################### # Value formatting ########################################################################### # Kernel values are parsed using locale-independent syntax. Locale is applied # when producing human-readable output. def initialise_locale() -> None: """Initialise LC_NUMERIC for display formatting only.""" try: locale.setlocale(locale.LC_NUMERIC, "") except locale.Error: locale.setlocale(locale.LC_NUMERIC, "C") def normalise_spacing(text: str) -> str: """Replace non-breaking locale grouping spaces with ordinary spaces. Some locales use U+00A0 (NO-BREAK SPACE) or U+202F (NARROW NO-BREAK SPACE) as thousands separators. Replace them with ordinary spaces to preserve terminal column alignment. """ return text.replace("\u00a0", " ").replace("\u202f", " ") def format_int(value: int) -> str: """Format an integer for display using LC_NUMERIC grouping.""" return normalise_spacing(locale.format_string("%d", value, grouping=True)) def format_decimal(value: float, digits: int = 1) -> str: """Format a decimal number using the current locale's decimal separator. Python's fixed-point formatting always uses '.', so the separator is replaced afterwards when the active locale uses a different character. """ text = f"{value:.{digits}f}" decimal_point = locale.localeconv().get("decimal_point") or "." if decimal_point != ".": text = text.replace(".", decimal_point) return normalise_spacing(text) def parse_int(text: str) -> int | None: """Parse an integer, returning None on failure.""" try: return int(text) except ValueError: return None def parse_float(text: str) -> float | None: """Parse a floating-point value, returning None on failure.""" try: return float(text) except ValueError: return None def format_bytes(raw: str) -> DisplayCell: """Format a byte count with binary units while preserving sort value.""" if raw == "max": return DisplayCell("max", raw, Align.RIGHT) value = parse_int(raw) if value is None: return DisplayCell(raw, raw, Align.RIGHT) units = ("bytes", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB",) amount = float(value) unit_index = 0 while amount >= 1024 and unit_index < len(units) - 1: amount /= 1024 unit_index += 1 if unit_index == 0: text = f"{format_int(value)} bytes" else: digits = 1 if amount >= 10 else 2 number = format_decimal(amount, digits).rstrip("0").rstrip(locale.localeconv()["decimal_point"]) text = f"{number} {units[unit_index]}" return DisplayCell(text, value, Align.RIGHT) def format_usec(raw: str) -> DisplayCell: """Format a microsecond count as H:MM:SS while preserving its sort value.""" value = parse_int(raw) if value is None: return DisplayCell(raw, raw, Align.RIGHT) seconds = value // 1_000_000 hours = seconds // 3600 minutes = (seconds % 3600) // 60 remaining = seconds % 60 return DisplayCell(f"{hours}:{minutes:02d}:{remaining:02d}", value, Align.RIGHT) def format_percent(value: float | None) -> DisplayCell: """Format a percentage value while preserving its sort value.""" if value is None: return DisplayCell("", None, Align.RIGHT) return DisplayCell(f"{format_decimal(value, 1)}%", value, Align.RIGHT) def format_cpu_max(raw: str | None) -> DisplayCell: """Format cpu.max quota and period as a percentage of one CPU.""" if not raw: return DisplayCell("", None, Align.RIGHT) tokens = raw.split() if len(tokens) != 2: return DisplayCell(raw, raw, Align.RIGHT) quota_text, period_text = tokens period = parse_int(period_text) if quota_text == "max": return DisplayCell("\u221e", float("inf"), Align.RIGHT) quota = parse_int(quota_text) if quota is None or period is None or period <= 0: return DisplayCell(raw, raw, Align.RIGHT) percent = (quota / period) * 100 text = f"{format_int(int(percent))}%" if percent.is_integer() else f"{format_decimal(percent, 1)}%" return DisplayCell(text, percent, Align.RIGHT) def format_pages(raw: str) -> DisplayCell: """Format a page count without assuming the host page size.""" value = parse_int(raw) if value is None: return DisplayCell(raw, raw, Align.RIGHT) return DisplayCell(f"{format_int(value)} pages", value, Align.RIGHT) def format_count(raw: str) -> DisplayCell: """Format an integer value while preserving special kernel tokens.""" if raw == "max": return DisplayCell("max", raw, Align.RIGHT) value = parse_int(raw) if value is None: return DisplayCell(raw, raw, Align.RIGHT) return DisplayCell(format_int(value), value, Align.RIGHT) def format_io_weight_default(text: str | None) -> DisplayCell: """Format the default io.weight value and indicate device-specific overrides.""" parsed = parse_flat_keyed(text) raw = parsed.get("default") if raw is None: return DisplayCell("", None, Align.RIGHT) cell = format_count(raw) if len(parsed) > 1: cell.text = f"{cell.text}*" return cell def format_cell(raw: str | None, unit: UnitKind) -> DisplayCell: """Format one cgroup value according to its unit metadata.""" if raw is None or raw == "": return DisplayCell("", None, Align.RIGHT) if unit == UnitKind.BYTES: return format_bytes(raw) if unit == UnitKind.USEC: return format_usec(raw) if unit == UnitKind.PAGES: return format_pages(raw) if unit in {UnitKind.COUNT, UnitKind.OPS}: return format_count(raw) if unit == UnitKind.PERCENT: if raw == "max": return DisplayCell("max", raw, Align.RIGHT) value = parse_float(raw) return format_percent(value) return DisplayCell(raw.replace("\n", "\\n"), raw, Align.LEFT) ########################################################################### # Cgroup hierarchy and filesystem access ########################################################################### # This section discovers and scans the cgroup v2 hierarchy, then performs # validated directory-relative interface I/O. def decode_mountinfo_path(value: str) -> str: """Decode octal escapes in a /proc/self/mountinfo path field.""" # mountinfo encodes spaces, tabs, newlines, and backslashes as octal. return value.replace("\\040", " ").replace("\\011", "\t").replace("\\012", "\n").replace("\\134", "\\") def discover_cgroup2_mount(mountinfo_path: Path = Path("/proc/self/mountinfo")) -> MountInfo: """Return the first cgroup v2 mount listed in a mountinfo file.""" try: lines = mountinfo_path.read_text(encoding="utf-8").splitlines() except OSError as exc: raise CgtreeError(f"cannot read {mountinfo_path}: {exc}") from exc for line in lines: pre, sep, post = line.partition(" - ") if not sep: continue pre_fields = pre.split() post_fields = post.split() if len(pre_fields) < 5 or len(post_fields) < 3: continue if post_fields[0] != "cgroup2": continue mount_point = Path(decode_mountinfo_path(pre_fields[4])) options = tuple(post_fields[2].split(",")) return MountInfo(mount_point=mount_point, options=options) raise CgtreeError("cgroup v2 mount not found") def scan_cgroup_tree(root_path: Path) -> CgroupNode: """Scan cgroup directories below root_path without crossing filesystems.""" try: root_device = root_path.stat().st_dev except OSError as exc: raise CgtreeError(f"cannot stat cgroup mount {root_path}: {exc}") from exc root = CgroupNode(path_abs=root_path, path_rel="", name="") def walk(parent: CgroupNode) -> None: try: entries = sorted(os.scandir(parent.path_abs), key=lambda entry: entry.name) except OSError: return for entry in entries: try: stat_result = entry.stat(follow_symlinks=False) except OSError: continue if not entry.is_dir(follow_symlinks=False) or stat_result.st_dev != root_device: continue relpath = entry.name if not parent.path_rel else f"{parent.path_rel}/{entry.name}" child = CgroupNode( path_abs=Path(entry.path), path_rel=relpath, name=entry.name, parent=parent, ) parent.children.append(child) walk(child) walk(root) return root def flatten_tree(root: CgroupNode) -> list[CgroupNode]: """Return all non-root cgroup nodes in pre-order traversal.""" nodes: list[CgroupNode] = [] def walk(node: CgroupNode) -> None: if node.path_rel: nodes.append(node) for child in node.children: walk(child) walk(root) return nodes def build_node_index(root: CgroupNode) -> dict[str, CgroupNode]: """Index scanned cgroups by path relative to the cgroup mount.""" return {node.path_rel: node for node in flatten_tree(root)} def find_node_by_relpath(root: CgroupNode, relpath: str) -> CgroupNode | None: """Return a scanned cgroup by path relative to the cgroup mount.""" return build_node_index(root).get(relpath) def is_valid_relative_cgroup_path(path_rel: str) -> bool: """Return whether path_rel is a safe canonical path relative to the mount.""" if not path_rel or path_rel.startswith("/") or "\x00" in path_rel: return False return all(component not in {"", ".", ".."} for component in path_rel.split("/")) def validate_relative_cgroup_path(path_rel: str, source: str) -> None: """Reject unsafe or non-canonical paths relative to the cgroup mount.""" if not is_valid_relative_cgroup_path(path_rel): raise CgtreeError(f"invalid cgroup path for {source}: {path_rel}") def select_cgroups( root: CgroupNode, node_index: dict[str, CgroupNode], matchers: Sequence[str], match_mode: str, ) -> list[CgroupNode]: """Resolve positional basename matchers or one exact relative path.""" if any("/" in matcher for matcher in matchers): if len(matchers) != 1: raise CgtreeError("an exact cgroup path must be the only positional MATCH") path_rel = matchers[0] validate_relative_cgroup_path(path_rel, "MATCH") node = node_index.get(path_rel) if node is None: raise CgtreeError(f"exact cgroup path does not exist: {path_rel}") return [node] return match_nodes(root, matchers, match_mode) def match_nodes( root: CgroupNode, matchers: Sequence[str], match_mode: str, ) -> list[CgroupNode]: """Find cgroups whose basename matches any or all substring matchers.""" if not matchers: return list(root.children) matches: list[CgroupNode] = [] for node in flatten_tree(root): checks = [matcher in node.name for matcher in matchers] if (match_mode == "all" and all(checks)) or (match_mode == "any" and any(checks)): matches.append(node) return matches INTERFACE_FILENAME_RE = re.compile(r"[A-Za-z0-9_.-]+", re.ASCII) def is_valid_interface_filename(file_name: str) -> bool: """Return whether file_name is a valid ASCII interface basename.""" return ( bool(file_name) and file_name not in {".", ".."} and file_name.isascii() and INTERFACE_FILENAME_RE.fullmatch(file_name) is not None ) def validate_interface_filename(file_name: str) -> None: """Reject interface names that are unsafe for directory-relative I/O.""" if not is_valid_interface_filename(file_name): raise CgtreeError(f"invalid cgroup interface filename: {file_name}") def open_cgroup_directory(node: CgroupNode) -> int: """Open a scanned cgroup directory without following a replacement symlink.""" flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW return os.open(node.path_abs, flags) def interface_exists(node: CgroupNode, file_name: str) -> bool: """Return whether a regular interface file exists directly in a cgroup.""" validate_interface_filename(file_name) try: directory_fd = open_cgroup_directory(node) except OSError: return False try: try: result = os.stat(file_name, dir_fd=directory_fd, follow_symlinks=False) except OSError: return False return stat.S_ISREG(result.st_mode) finally: os.close(directory_fd) def interface_names_for_node(node: CgroupNode) -> set[str]: """Return valid regular interface filenames directly in one cgroup.""" names: set[str] = set() try: entries = os.scandir(node.path_abs) except OSError: return names with entries: for entry in entries: if not is_valid_interface_filename(entry.name): continue try: if entry.is_file(follow_symlinks=False): names.add(entry.name) except OSError: continue return names def existing_interface_names(root: CgroupNode) -> set[str]: """Return the union of interface filenames present in the scanned hierarchy.""" names = interface_names_for_node(root) for node in flatten_tree(root): names.update(interface_names_for_node(node)) return names def read_interface_text(node: CgroupNode, file_name: str) -> str: """Read a validated interface file relative to its cgroup directory.""" validate_interface_filename(file_name) directory_fd = open_cgroup_directory(node) interface_fd = -1 try: flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW interface_fd = os.open(file_name, flags, dir_fd=directory_fd) handle = os.fdopen(interface_fd, "r", encoding="utf-8") interface_fd = -1 with handle: return handle.read().strip() finally: if interface_fd >= 0: os.close(interface_fd) os.close(directory_fd) def write_interface_text(node: CgroupNode, file_name: str, value: str) -> None: """Write a validated interface file relative to its cgroup directory.""" validate_interface_filename(file_name) directory_fd = -1 interface_fd = -1 try: directory_fd = open_cgroup_directory(node) flags = os.O_WRONLY | os.O_TRUNC | os.O_CLOEXEC | os.O_NOFOLLOW interface_fd = os.open(file_name, flags, dir_fd=directory_fd) handle = os.fdopen(interface_fd, "w", encoding="utf-8") interface_fd = -1 with handle: handle.write(f"{value}\n") except OSError as exc: cgroup_name = node.path_rel or "/" raise CgtreeError(f"cannot write {file_name} in cgroup {cgroup_name}: {exc}") from exc finally: if interface_fd >= 0: os.close(interface_fd) if directory_fd >= 0: os.close(directory_fd) def descendants_to_depth(root: CgroupNode, depth: int | None) -> list[CgroupNode]: """Return root and descendants through depth, or all descendants if unlimited.""" result: list[CgroupNode] = [] def walk(node: CgroupNode, current_depth: int) -> None: result.append(node) if depth is not None and current_depth >= depth: return for child in node.children: walk(child, current_depth + 1) walk(root, 0) return result def visible_nodes( root: CgroupNode, display_roots: Sequence[CgroupNode], depth: int | None, include_context: bool, ) -> tuple[list[CgroupNode], set[str]]: """Return visible nodes and the subset that should contain data values.""" # data_relpaths marks rows that receive values. Ancestor context rows remain # visible but empty so matched descendants retain their place in the hierarchy. visible_by_rel: dict[str, CgroupNode] = {} data_relpaths: set[str] = set() for display_root in display_roots: for node in descendants_to_depth(display_root, depth): visible_by_rel[node.path_rel] = node data_relpaths.add(node.path_rel) if include_context: for display_root in display_roots: current = display_root.parent while current is not None and current.path_rel: visible_by_rel[current.path_rel] = current current = current.parent ordered = [node for node in flatten_tree(root) if node.path_rel in visible_by_rel] return ordered, data_relpaths ####### Interface reads and parsers ######## # Interface values may change while a table is being built. FileCache gives one # display or sort pass a consistent view. class FileCache: """Cache interface reads for one display or sort pass. Cgroup interface files can change between reads. Caching ensures that every selector in one pass sees the same value for a given interface. """ def __init__(self) -> None: """Initialise an empty per-pass cache.""" # Keys are (relative cgroup path, interface name). self._values: dict[tuple[str, str], str | None] = {} def read(self, node: CgroupNode, file_name: str) -> str | None: """Read one interface once and return the cached value thereafter.""" key = (node.path_rel, file_name) if key not in self._values: try: self._values[key] = read_interface_text(node, file_name) except OSError: # Some interfaces are absent or unreadable for particular cgroup types. # Cache None so every selector observes the same result. self._values[key] = None return self._values[key] def parse_flat_keyed(text: str | None) -> dict[str, str]: """Parse a flat-keyed interface into one mapping. Most flat-keyed interfaces use one KEY VALUE pair per line. HugeTLB NUMA statistics instead use one line of KEY=VALUE tokens, such as "total=0 N0=0 N1=0". Both layouts represent flat mappings. """ result: dict[str, str] = {} if not text: return result for line in text.splitlines(): tokens = line.split() if not tokens: continue # Inline KEY=VALUE form used by hugetlb..numa_stat. Requiring '=' in the # first token avoids changing the interpretation of ordinary KEY VALUE rows. if "=" in tokens[0]: for token in tokens: key, separator, value = token.partition("=") if separator and key: result[key] = value continue parts = line.split(maxsplit=1) if len(parts) == 2: result[parts[0]] = parts[1] return result def parse_nested_keyed(text: str | None) -> dict[str, dict[str, str]]: """Parse a nested-keyed interface into keyed field mappings.""" # io.stat rows look like "8:0 rbytes=1 wbytes=2". Keep the row key # separate from its FIELD=VALUE entries for later expansion. result: dict[str, dict[str, str]] = {} if not text: return result for line in text.splitlines(): tokens = line.split() if not tokens: continue key = tokens[0] values: dict[str, str] = {} for token in tokens[1:]: field_name, sep, value = token.partition("=") if sep: values[field_name] = value elif len(tokens) == 2: # DMEM uses REGION VALUE rather than FIELD=VALUE, so expose the scalar as a # synthetic field named "value". values["value"] = token result[key] = values return result def parse_list(text: str | None) -> list[str]: """Parse whitespace-separated cgroup list files.""" if not text: return [] return text.split() def get_file_spec(file_name: str) -> FileSpec: """Return metadata for an interface, using conservative defaults if unknown.""" if file_name in FILE_SPECS: return FILE_SPECS[file_name] hugetlb_kind = hugetlb_interface_kind(file_name) if hugetlb_kind in {"events", "events.local"}: return FileSpec( name=file_name, interface_format=InterfaceFormat.FLAT, default_unit=UnitKind.COUNT, ) if hugetlb_kind == "numa_stat": # HugeTLB NUMA statistics use one line such as "total=0 N0=0 N1=0". Treat # the line as a flat mapping so every node present in the file is exposed # without assuming the host's NUMA topology. return FileSpec( name=file_name, interface_format=InterfaceFormat.FLAT, default_unit=UnitKind.BYTES, ) if hugetlb_kind in {"current", "max", "rsvd.current", "rsvd.max"}: return FileSpec( name=file_name, interface_format=InterfaceFormat.SCALAR, default_unit=UnitKind.BYTES, writable=hugetlb_kind in {"max", "rsvd.max"}, ) # Display unknown kernel interfaces as raw text rather than guessing their # format or units. return FileSpec(name=file_name, interface_format=InterfaceFormat.SCALAR) def is_known_file_name(file_name: str) -> bool: """Return whether cgtree has static or dynamic metadata for file_name.""" return file_name in FILE_SPECS or hugetlb_interface_kind(file_name) is not None def unit_for_field(file_name: str, field_name: str | None) -> UnitKind: """Return the display unit for an interface or one of its fields.""" spec = get_file_spec(file_name) if field_name is None: return spec.default_unit # NUMA node field names are generated by the kernel (N0, N1, ...). # Their values are byte counts for both memory.numa_stat and # hugetlb..numa_stat. if spec.default_unit == UnitKind.BYTES and ( file_name == "memory.numa_stat" or hugetlb_interface_kind(file_name) == "numa_stat" ): return UnitKind.BYTES # Keep unknown fields as raw text. return spec.field_units.get(field_name, UnitKind.STRING) def read_scalar_value(cache: FileCache, node: CgroupNode, file_name: str) -> str | None: """Read the first line from a scalar cgroup interface.""" text = cache.read(node, file_name) if text is None: return None return text.splitlines()[0] if text.splitlines() else "" def read_flat_field( cache: FileCache, node: CgroupNode, file_name: str, field_name: str, ) -> str | None: """Read one field from a flat-keyed cgroup interface.""" return parse_flat_keyed(cache.read(node, file_name)).get(field_name) def read_nested_field_values( cache: FileCache, node: CgroupNode, file_name: str, field_name: str, ) -> dict[str, str]: """Read one field across all keys in a nested-keyed cgroup file.""" parsed = parse_nested_keyed(cache.read(node, file_name)) return {key: values[field_name] for key, values in parsed.items() if field_name in values} def aggregate_nested_field( cache: FileCache, node: CgroupNode, file_name: str, field_name: str, ) -> str | None: """Sum one numeric field across all keys in a nested-keyed file.""" values = read_nested_field_values(cache, node, file_name, field_name) if not values: return None total = 0 for value in values.values(): parsed = parse_int(value) if parsed is None: return None total += parsed return str(total) def read_nested_key_field( cache: FileCache, node: CgroupNode, file_name: str, selector: str, ) -> str | None: """Read a nested KEY.FIELD selector such as some.avg10.""" key, sep, field_name = selector.partition(".") if not sep or not key or not field_name: return None return parse_nested_keyed(cache.read(node, file_name)).get(key, {}).get(field_name) def read_nested_key_row( cache: FileCache, node: CgroupNode, file_name: str, key_name: str, ) -> str | None: """Read the raw remainder of one exact row from a nested-keyed file.""" text = cache.read(node, file_name) if not text: return None for line in text.splitlines(): key, separator, remainder = line.partition(" ") if key == key_name: return remainder if separator else "" return None ########################################################################### # Selector handling ########################################################################### # A selector is an expression accepted by -C, such as memory.current, # memory.stat.anon, or irq.pressure.full.avg10. This section resolves selectors # into interface, key, field, unit, and display metadata. def split_known_file_field(name: str) -> tuple[str, str | None]: """Split a selector into a static interface name and optional field.""" if name in FILE_SPECS: return name, None for file_name in sorted(FILE_SPECS, key=len, reverse=True): prefix = f"{file_name}." if name.startswith(prefix): return file_name, name[len(prefix) :] return name, None def split_existing_file_selector( name: str, available_interface_files: set[str], ) -> tuple[str, str | None]: """Resolve a selector against interfaces present in the scanned hierarchy. Unknown interfaces may be selected only as complete raw files. Field and row selectors are interpreted only for known interfaces. """ if name in available_interface_files: return name, None candidates = sorted(available_interface_files, key=len, reverse=True) for file_name in candidates: if not is_known_file_name(file_name): continue spec = get_file_spec(file_name) if spec.interface_format not in {InterfaceFormat.FLAT, InterfaceFormat.NESTED}: continue prefix = f"{file_name}." if name.startswith(prefix) and name[len(prefix) :]: return file_name, name[len(prefix) :] raise CgtreeError(f"unknown or unavailable cgroup interface selector: {name}") def known_nested_field_names(file_name: str, spec: FileSpec) -> set[str]: """Return nested field names described by cgtree metadata.""" return set(spec.field_units) | set(nested_display_fields(file_name)) def explicit_selector_tokens(text: str | None) -> list[str]: """Split a comma-separated -C value into non-empty selectors.""" if not text: return [] return [part.strip() for part in text.split(",") if part.strip()] def make_selector_spec( name: str, pretty: bool, aggregate_io: bool = False, available_interface_files: set[str] | None = None, ) -> SelectorSpec: """Resolve a selector into the metadata required for reading and display.""" if name == "cgroup.procs.count": return SelectorSpec( name=name, header=PRETTY_HEADERS[name] if pretty else name, file_name="cgroup.procs", unit=UnitKind.COUNT, virtual=True, header_lines=(PRETTY_HEADERS[name] if pretty else name,), ) if name == "cpu.percent": return SelectorSpec( name=name, header=PRETTY_HEADERS[name] if pretty else name, unit=UnitKind.PERCENT, virtual=True, header_lines=(PRETTY_HEADERS[name] if pretty else name,), ) if name == "cpu.max.percent": return SelectorSpec( name=name, header=PRETTY_HEADERS[name] if pretty else name, file_name="cpu.max", virtual=True, header_lines=(PRETTY_HEADERS[name] if pretty else name,), ) if name == "io.weight.default": return SelectorSpec( name=name, header=PRETTY_HEADERS[name] if pretty else name, file_name="io.weight", unit=UnitKind.COUNT, virtual=True, header_lines=(PRETTY_HEADERS[name] if pretty else name,), ) if available_interface_files is None: file_name, selector = split_known_file_field(name) else: file_name, selector = split_existing_file_selector(name, available_interface_files) spec = get_file_spec(file_name) key_name: str | None = None field_name: str | None = None if selector is not None: if spec.interface_format == InterfaceFormat.FLAT: field_name = selector elif spec.interface_format == InterfaceFormat.NESTED: nested_fields = known_nested_field_names(file_name, spec) # The kernel generates memory.numa_stat fields as N0, N1, and so on. Recognise # any N suffix. candidate_field = selector.rsplit(".", maxsplit=1)[-1] if file_name == "memory.numa_stat" and NUMA_NODE_FIELD_RE.fullmatch(candidate_field): field_name = candidate_field key_prefix = selector[: -(len(candidate_field) + 1)] if "." in selector else "" key_name = key_prefix or None else: field_matches = sorted( (field for field in nested_fields if selector.endswith(f".{field}")), key=len, reverse=True, ) if field_matches: field_name = field_matches[0] key_name = selector[: -(len(field_name) + 1)] elif selector in nested_fields: field_name = selector else: # An unrecognised selector on a nested interface is treated as an exact row # key and displayed as raw text. key_name = selector else: raise CgtreeError(f"interface does not support field selectors: {file_name}") full_file = selector is None and spec.interface_format in {InterfaceFormat.FLAT, InterfaceFormat.NESTED} unit = UnitKind.STRING if key_name is not None and field_name is None else unit_for_field(file_name, field_name) header = PRETTY_HEADERS.get(name, name) if pretty else name header_lines = (header,) if pretty or selector is None else (file_name, selector) return SelectorSpec( name=name, header=header, file_name=file_name, key_name=key_name, field_name=field_name, unit=unit, full_file=full_file, aggregate=( aggregate_io and file_name == "io.stat" and key_name is None and field_name is not None ), virtual=False, header_lines=header_lines, ) def resolve_selectors( args: argparse.Namespace, available_interface_files: set[str], ) -> tuple[list[SelectorSpec], bool]: """Resolve explicit -C selectors or the requested built-in views.""" if args.columns: tokens = explicit_selector_tokens(args.columns) return [ make_selector_spec( token, pretty=False, available_interface_files=available_interface_files, ) for token in tokens ], False pretty = True tokens: list[str] = [] if args.memory: tokens.extend(MEMORY_VIEW_SELECTORS) if args.cpu: cpu_selectors = list(CPU_VIEW_SELECTORS) if args.no_sample: cpu_selectors.remove("cpu.percent") tokens.extend(cpu_selectors) if args.io: tokens.extend(IO_VIEW_SELECTORS) if args.show_pids: tokens.extend(PIDS_VIEW_SELECTORS) if not tokens: tokens.extend(DEFAULT_VIEW_SELECTORS) return [make_selector_spec(token, pretty=pretty, aggregate_io=True) for token in tokens], pretty def list_available_interface_files(root: CgroupNode) -> list[str]: """Return interface filenames present in the scanned hierarchy.""" return sorted(existing_interface_names(root)) ########################################################################### # Runtime data collection ########################################################################### # These helpers collect values that require sampling or external lookups. ####### CPU sampling ######## # Sampled CPU percentage is a virtual selector derived from two cpu.stat reads. def read_cpu_times(cache: FileCache, node: CgroupNode) -> dict[str, int]: """Read CPU usage counters from cpu.stat for one cgroup.""" values = parse_flat_keyed(cache.read(node, "cpu.stat")) result: dict[str, int] = {} for key in ("usage_usec", "user_usec", "system_usec"): parsed = parse_int(values.get(key, "")) if parsed is not None: result[key] = parsed return result def sample_cpu_percent(nodes: Sequence[CgroupNode], period: float) -> dict[str, dict[str, float]]: """Sample cpu.stat twice and return CPU percentages by relative cgroup path.""" first_cache = FileCache() first = {node.path_rel: read_cpu_times(first_cache, node) for node in nodes} time.sleep(period) second_cache = FileCache() sampled: dict[str, dict[str, float]] = {} denominator = period * 1_000_000 if denominator <= 0: return sampled for node in nodes: before = first.get(node.path_rel, {}) after = read_cpu_times(second_cache, node) row: dict[str, float] = {} for source_key, output_key in ( ("usage_usec", "cpu.percent"), ("user_usec", "cpu.user_percent"), ("system_usec", "cpu.system_percent"), ): if source_key not in before or source_key not in after: continue delta = max(after[source_key] - before[source_key], 0) row[output_key] = (delta / denominator) * 100 sampled[node.path_rel] = row return sampled ####### Device and process helpers ######## # Device-keyed interfaces use kernel major:minor identifiers. These helpers # resolve them for display or writes and optionally read process names for # --comm rows. def resolve_block_device(major_minor: str) -> str | None: """Resolve a kernel major:minor block key to a device name when possible.""" path = Path("/sys/dev/block") / major_minor try: return path.resolve().name except OSError: return None def block_name_to_major_minor(name: str) -> str | None: """Resolve a block device name to the major:minor key required for writes.""" path = Path("/sys/class/block") / name / "dev" try: return path.read_text(encoding="utf-8").strip() except OSError: return None def display_device_key(key: str, dev_format: str) -> tuple[str, bool]: """Format a device key and indicate whether it belongs in the DEV column.""" name = resolve_block_device(key) if dev_format == "kernel" or name is None: return key, False if dev_format == "both": return f"{key}/{name}", True return name, True def normalise_device_key(key: str) -> str: """Convert a block-device name or major:minor key to the kernel form.""" if re.fullmatch(r"\d+:\d+", key): return key resolved = block_name_to_major_minor(key) if resolved is None: raise CgtreeError(f"cannot resolve block device name: {key}") return resolved def read_process_comm(pid: int, proc_root: Path = Path("/proc")) -> ProcessInfo: """Read a process name from /proc/PID/comm. Processes may exit or become inaccessible between reading cgroup.procs and this lookup. In that case, retain the PID and leave the command name empty. """ try: comm = (proc_root / str(pid) / "comm").read_text( encoding="utf-8", errors="replace", ).rstrip("\n") except OSError: return ProcessInfo(pid=pid) return ProcessInfo(pid=pid, comm=comm) def process_infos_for_node(cache: FileCache, node: CgroupNode) -> list[ProcessInfo]: """Return process details for unique PIDs listed in cgroup.procs.""" pids = {int(value) for value in parse_list(cache.read(node, "cgroup.procs")) if value.isdecimal()} return [read_process_comm(pid) for pid in sorted(pids)] ########################################################################### # Display generation ########################################################################### # Resolved selectors are converted into scalar cells or expanded KEY/DEV rows. # Tree labels and continuation rows are built here before the final plain-text # renderer calculates column widths. def node_selector_cell( cache: FileCache, node: CgroupNode, selector: SelectorSpec, sampled_cpu: dict[str, dict[str, float]], ) -> DisplayCell: """Build the display cell for one cgroup and resolved selector.""" if selector.name == "cgroup.procs.count": pids = {value for value in parse_list(cache.read(node, "cgroup.procs")) if value.isdecimal()} return format_count(str(len(pids))) if selector.name == "cpu.percent": return format_percent(sampled_cpu.get(node.path_rel, {}).get(selector.name)) if selector.name == "cpu.max.percent": return format_cpu_max(cache.read(node, "cpu.max")) if selector.name == "io.weight.default": return format_io_weight_default(cache.read(node, "io.weight")) if selector.file_name is None: return DisplayCell("", None, Align.RIGHT) spec = get_file_spec(selector.file_name) raw: str | None if selector.key_name is not None: if spec.interface_format != InterfaceFormat.NESTED: raw = None elif selector.field_name is None: raw = read_nested_key_row(cache, node, selector.file_name, selector.key_name) else: raw = parse_nested_keyed(cache.read(node, selector.file_name)).get( selector.key_name, {}, ).get(selector.field_name) elif selector.field_name is None: if selector.file_name in {"io.weight", "io.bfq.weight"}: raw = parse_flat_keyed(cache.read(node, selector.file_name)).get("default") elif spec.interface_format == InterfaceFormat.LIST: raw = " ".join(parse_list(cache.read(node, selector.file_name))) elif selector.file_name not in FILE_SPECS: raw = cache.read(node, selector.file_name) else: raw = read_scalar_value(cache, node, selector.file_name) elif spec.interface_format == InterfaceFormat.FLAT: raw = read_flat_field(cache, node, selector.file_name, selector.field_name) elif spec.interface_format == InterfaceFormat.NESTED: if selector.aggregate: raw = aggregate_nested_field(cache, node, selector.file_name, selector.field_name) else: values = read_nested_field_values(cache, node, selector.file_name, selector.field_name) raw = " ".join(f"{key}={value}" for key, value in sorted(values.items())) or None else: raw = None return format_cell(raw, selector.unit) def is_fixed_nested_key(selector: SelectorSpec) -> bool: """Return true for a nested selector naming one complete keyed row.""" if selector.file_name is None or selector.key_name is None or selector.field_name is not None: return False return get_file_spec(selector.file_name).interface_format == InterfaceFormat.NESTED def discover_nested_display_fields( cache: FileCache, nodes: Sequence[CgroupNode], data_relpaths: set[str], selectors: Sequence[SelectorSpec], ) -> dict[str, tuple[str, ...]]: """Discover runtime-generated fields required for nested-interface headers. memory.numa_stat contains one row per memory type and one N field per memory-bearing NUMA node. Reading the selected interface files avoids assuming contiguous node IDs or relying on separate topology data. The shared FileCache reuses those reads later in the same display pass. """ files_to_scan = { selector.file_name for selector in selectors if selector.file_name is not None and (selector.full_file or is_fixed_nested_key(selector)) and (metadata := NESTED_INTERFACES.get(selector.file_name)) is not None and metadata.discover_display_fields } discovered: dict[str, tuple[str, ...]] = {} for file_name in files_to_scan: fields: set[str] = set() for node in nodes: if node.path_rel not in data_relpaths: continue parsed = parse_nested_keyed(cache.read(node, file_name)) for values in parsed.values(): fields.update(name for name in values if NUMA_NODE_FIELD_RE.fullmatch(name)) if fields: discovered[file_name] = tuple( sorted(fields, key=lambda name: int(name[1:])) ) return discovered def expanded_column_headers( selectors: Sequence[SelectorSpec], discovered_fields: dict[str, tuple[str, ...]] | None = None, ) -> tuple[list[str], dict[str, tuple[str, ...]]]: """Build internal column names and one- or two-line display headings.""" headers: list[str] = [] header_lines: dict[str, tuple[str, ...]] = {} for selector in selectors: spec = get_file_spec(selector.file_name or "") if is_fixed_nested_key(selector): assert selector.key_name is not None for field_name in nested_display_fields(spec.name, discovered_fields): header = f"{spec.name}.{selector.key_name}.{field_name}" headers.append(header) header_lines[header] = (spec.name, f"{selector.key_name}.{field_name}") continue if not selector.full_file: headers.append(selector.header) header_lines[selector.header] = selector.header_lines or (selector.header,) continue if spec.interface_format == InterfaceFormat.NESTED: for field_name in nested_display_fields(spec.name, discovered_fields): header = f"{spec.name}.{field_name}" headers.append(header) header_lines[header] = (spec.name, field_name) else: headers.append(selector.header) header_lines[selector.header] = selector.header_lines or (selector.header,) return headers, header_lines def fixed_nested_key_cells( cache: FileCache, node: CgroupNode, selector: SelectorSpec, discovered_fields: dict[str, tuple[str, ...]] | None = None, ) -> dict[str, DisplayCell]: """Format the displayed fields for one selected nested row.""" if selector.file_name is None or selector.key_name is None: return {} spec = get_file_spec(selector.file_name) values = parse_nested_keyed(cache.read(node, selector.file_name)).get(selector.key_name, {}) return { f"{spec.name}.{selector.key_name}.{field_name}": format_cell( values.get(field_name), unit_for_field(spec.name, field_name), ) for field_name in nested_display_fields(spec.name, discovered_fields) } def expanded_entries_for_node( cache: FileCache, node: CgroupNode, selectors: Sequence[SelectorSpec], options: DisplayOptions, discovered_fields: dict[str, tuple[str, ...]] | None = None, ) -> tuple[list[tuple[str, str, str, bool]], dict[tuple[str, str], DisplayCell]]: """Build KEY/DEV continuation rows and their cells for one cgroup.""" entries: dict[tuple[str, str], tuple[str, str, str, bool]] = {} cells: dict[tuple[str, str], DisplayCell] = {} for selector in selectors: if not selector.full_file and not is_expanded_nested_field(selector): continue if selector.file_name is None: continue spec = get_file_spec(selector.file_name) if spec.interface_format == InterfaceFormat.FLAT: parsed = parse_flat_keyed(cache.read(node, selector.file_name)) for key, raw in parsed.items(): identity = ("dev" if spec.device_keyed and key != "default" else "key", key) label, is_device = device_label_for_key(key, spec.device_keyed, options.dev_format) entries[identity] = (identity[0], key, label, is_device) cells[(selector.header, key)] = format_cell(raw, unit_for_field(spec.name, key)) elif spec.interface_format == InterfaceFormat.NESTED: parsed = parse_nested_keyed(cache.read(node, selector.file_name)) requested_fields = nested_requested_fields(selector, spec, discovered_fields) for key, values in parsed.items(): label, is_device = device_label_for_key(key, spec.device_keyed, options.dev_format) identity = ("dev" if is_device else "key", key) entries[identity] = (identity[0], key, label, is_device) for field_name in requested_fields: header = selector.header if selector.field_name is not None else f"{spec.name}.{field_name}" cells[(header, key)] = format_cell( values.get(field_name), unit_for_field(spec.name, field_name), ) ordered_entries = [entries[key] for key in sorted(entries)] return ordered_entries, cells def device_label_for_key(key: str, device_keyed: bool, dev_format: str) -> tuple[str, bool]: """Return the display label for a keyed row and whether it is a DEV value.""" if not device_keyed or key == "default": return key, False return display_device_key(key, dev_format) def is_expanded_nested_field(selector: SelectorSpec) -> bool: """Return true when one nested field should be shown per KEY/DEV row.""" if ( selector.aggregate or selector.file_name is None or selector.key_name is not None or selector.field_name is None ): return False return get_file_spec(selector.file_name).interface_format == InterfaceFormat.NESTED def nested_requested_fields( selector: SelectorSpec, spec: FileSpec, discovered_fields: dict[str, tuple[str, ...]] | None = None, ) -> tuple[str, ...]: """Return the nested fields displayed for one selector.""" if selector.field_name is not None: return (selector.field_name,) return nested_display_fields(spec.name, discovered_fields) def compact_value_for_selector(cache: FileCache, node: CgroupNode, selector: SelectorSpec) -> DisplayCell: """Format a multi-row interface as one deterministic compact value.""" if selector.file_name is None: return DisplayCell("", None, Align.RIGHT) spec = get_file_spec(selector.file_name) if spec.interface_format == InterfaceFormat.FLAT: parsed = parse_flat_keyed(cache.read(node, selector.file_name)) text = " ".join(f"{key}={value}" for key, value in sorted(parsed.items())) return DisplayCell(text, text, Align.LEFT) if spec.interface_format == InterfaceFormat.NESTED: parsed = parse_nested_keyed(cache.read(node, selector.file_name)) if selector.field_name is not None and selector.key_name is None: parts = [ f"{key}={values[selector.field_name]}" for key, values in sorted(parsed.items()) if selector.field_name in values ] text = " ".join(parts) return DisplayCell(text, text, Align.LEFT) parts: list[str] = [] for key, values in sorted(parsed.items()): fields = " ".join(f"{name}={value}" for name, value in sorted(values.items())) parts.append(f"{key} {fields}".strip()) text = "; ".join(parts) return DisplayCell(text, text, Align.LEFT) return DisplayCell("", None, Align.LEFT) def build_display_rows( root: CgroupNode, nodes: Sequence[CgroupNode], data_relpaths: set[str], selectors: Sequence[SelectorSpec], options: DisplayOptions, sampled_cpu: dict[str, dict[str, float]], show_comm: bool = False, ) -> tuple[list[str], dict[str, tuple[str, ...]], list[DisplayRow]]: """Build final table headers and rows from visible cgroups and selectors.""" cache = FileCache() visible_set = {node.path_rel for node in nodes} headers = ["CGROUP"] header_lines: dict[str, tuple[str, ...]] = {"CGROUP": ("CGROUP",)} discovered_fields = discover_nested_display_fields( cache, nodes, data_relpaths, selectors, ) value_headers, value_header_lines = expanded_column_headers( selectors, discovered_fields, ) rows: list[DisplayRow] = [] needs_key = False needs_dev = False process_infos_by_rel = ( { node.path_rel: process_infos_for_node(cache, node) for node in nodes if node.path_rel in data_relpaths } if show_comm else {} ) branch_down_relpaths = { node.path_rel for node in nodes if process_infos_by_rel.get(node.path_rel) and has_visible_descendant(node, visible_set) } for node in nodes: process_infos = process_infos_by_rel.get(node.path_rel, []) label = tree_label(node, visible_set, branch_down_relpaths) if node.path_rel not in data_relpaths: rows.append(DisplayRow(label=label, cells={}, is_context=True)) continue if options.compact: cells = {} for selector in selectors: if selector.full_file or is_expanded_nested_field(selector): cells[selector.header] = compact_value_for_selector(cache, node, selector) else: cells[selector.header] = node_selector_cell(cache, node, selector, sampled_cpu) rows.append(DisplayRow(label=label, cells=cells)) else: entries, expanded_cells = expanded_entries_for_node( cache, node, selectors, options, discovered_fields, ) scalar_cells = { selector.header: node_selector_cell(cache, node, selector, sampled_cpu) for selector in selectors if ( not selector.full_file and not is_expanded_nested_field(selector) and not is_fixed_nested_key(selector) ) } for selector in selectors: if is_fixed_nested_key(selector): scalar_cells.update( fixed_nested_key_cells( cache, node, selector, discovered_fields, ) ) if not entries: rows.append(DisplayRow(label=label, cells=scalar_cells)) else: for index, (kind, key, display_key, is_device) in enumerate(entries): row_cells = dict(scalar_cells) if index == 0 else {} for header in value_headers: row_cells[header] = expanded_cells.get((header, key), DisplayCell()) row_label = label if index == 0 else tree_continuation_label( node, visible_set, branch_down_relpaths, ) needs_dev = needs_dev or is_device needs_key = needs_key or not is_device or kind == "key" rows.append( DisplayRow( label=row_label, key="" if is_device else display_key, dev=display_key if is_device else "", cells=row_cells, ) ) if show_comm: continuation = tree_continuation_label( node, visible_set, branch_down_relpaths, ) for process in process_infos: rows.append( DisplayRow( label=continuation, cells={ "PID": DisplayCell(str(process.pid), process.pid, Align.RIGHT), "COMM": DisplayCell(process.comm, process.comm, Align.LEFT), }, ) ) if needs_key: headers.append("KEY") header_lines["KEY"] = ("KEY",) if needs_dev: headers.append("DEV") header_lines["DEV"] = ("DEV",) headers.extend(value_headers) header_lines.update(value_header_lines) if show_comm: headers.extend(("PID", "COMM")) header_lines["PID"] = ("PID",) header_lines["COMM"] = ("COMM",) _ = root return headers, header_lines, rows def tree_label( node: CgroupNode, visible_set: set[str], branch_down_relpaths: set[str] | None = None, ) -> str: """Return the visible tree label for the primary row of a cgroup.""" prefix, node_branch_rendered = tree_ancestor_prefix( node, visible_set, branch_down_relpaths, ) if node_branch_rendered: return prefix + node.name branch = TREE_BRANCH if has_visible_later_sibling(node, visible_set) else TREE_LAST connector = TREE_DOWN if node.path_rel in (branch_down_relpaths or set()) else TREE_HORIZONTAL return prefix + f"{branch}{connector} " + node.name def tree_continuation_label( node: CgroupNode, visible_set: set[str], branch_down_relpaths: set[str] | None = None, ) -> str: """Return the continuation tree label for an expanded row.""" prefix, _ = tree_ancestor_prefix( node, visible_set, branch_down_relpaths, ) sibling_guide = has_visible_later_sibling(node, visible_set) if node.path_rel in (branch_down_relpaths or set()): segment = ( f"{TREE_VERTICAL}{TREE_VERTICAL} " if sibling_guide else f" {TREE_VERTICAL} " ) else: segment = f"{TREE_VERTICAL} " if sibling_guide else " " return (prefix + segment).rstrip() def tree_ancestor_prefix( node: CgroupNode, visible_set: set[str], branch_down_relpaths: set[str] | None = None, ) -> tuple[str, bool]: """Build the ancestor guide prefix for one cgroup row. When a parent has process-detail rows before its visible children, its "┬" connector must continue through those rows. In that case the direct child's branch is included in the parent segment. The returned boolean indicates whether the current node's branch has already been emitted. """ ancestors: list[CgroupNode] = [] current = node.parent while current is not None and current.path_rel: ancestors.append(current) current = current.parent ancestors.reverse() branch_down_relpaths = branch_down_relpaths or set() parts: list[str] = [] node_branch_rendered = False for index, ancestor in enumerate(ancestors): next_node = ancestors[index + 1] if index + 1 < len(ancestors) else node if ancestor.path_rel in branch_down_relpaths: sibling_guide = ( TREE_VERTICAL if has_visible_later_sibling(ancestor, visible_set) else " " ) child_branch = ( TREE_BRANCH if has_visible_later_sibling(next_node, visible_set) else TREE_LAST ) parts.append(f"{sibling_guide}{child_branch}{TREE_HORIZONTAL} ") if next_node is node: node_branch_rendered = True else: parts.append( f"{TREE_VERTICAL} " if has_visible_later_sibling(ancestor, visible_set) else " " ) return "".join(parts), node_branch_rendered def has_visible_later_sibling(node: CgroupNode, visible_set: set[str]) -> bool: """Return true when tree guides must continue past this node.""" if node.parent is None: return False siblings = node.parent.children try: index = siblings.index(node) except ValueError: return False return any(subtree_contains_visible(sibling, visible_set) for sibling in siblings[index + 1 :]) def has_visible_descendant(node: CgroupNode, visible_set: set[str]) -> bool: """Return whether a node has a visible descendant.""" return any(subtree_contains_visible(child, visible_set) for child in node.children) def subtree_contains_visible(node: CgroupNode, visible_set: set[str]) -> bool: """Return true if node or one of its descendants is currently rendered.""" if node.path_rel in visible_set: return True return any(subtree_contains_visible(child, visible_set) for child in node.children) ####### Sorting ######## # Sorting preserves the cgroup hierarchy. Display headings and raw # selectors may both be used as sort columns. def normalise_sort_name(name: str) -> str: """Resolve a friendly column heading to its selector name.""" return HEADER_ALIASES.get(name, HEADER_ALIASES.get(name.upper(), name)) def selector_for_sort_name( name: str, selectors: Sequence[SelectorSpec], available_interface_files: set[str], ) -> SelectorSpec | None: """Resolve a sort column from a display heading or raw selector.""" internal = normalise_sort_name(name) for selector in selectors: if selector.name == internal or selector.header == name: return selector if internal == "CGROUP": return None return make_selector_spec( internal, pretty=False, aggregate_io=True, available_interface_files=available_interface_files, ) def sort_value_for_node( node: CgroupNode, sort_name: str, selectors: Sequence[SelectorSpec], sampled_cpu: dict[str, dict[str, float]], cache: FileCache, available_interface_files: set[str], ) -> DisplayCell: """Return the display cell used to sort one cgroup by one column.""" internal = normalise_sort_name(sort_name) if internal == "CGROUP": return DisplayCell(node.name, node.name, Align.LEFT) selector = selector_for_sort_name(sort_name, selectors, available_interface_files) if selector is None: return DisplayCell(node.name, node.name, Align.LEFT) return node_selector_cell(cache, node, selector, sampled_cpu) def compare_sort_values(left: DisplayCell, right: DisplayCell, reverse_natural: bool) -> int: """Compare sort cells while keeping empty values last.""" left_empty = left.sort_value is None or left.text == "" right_empty = right.sort_value is None or right.text == "" if left_empty and right_empty: return 0 if left_empty: return 1 if right_empty: return -1 left_value = left.sort_value right_value = right.sort_value if left_value == right_value: return 0 # Numeric columns sort highest first by default. Text columns sort # alphabetically. if isinstance(left_value, int | float) and isinstance(right_value, int | float): result = -1 if left_value > right_value else 1 return -result if reverse_natural else result result = -1 if str(left_value) < str(right_value) else 1 return -result if reverse_natural else result def sort_tree( root: CgroupNode, sort_names: Sequence[str], reverse_natural: bool, selectors: Sequence[SelectorSpec], sampled_cpu: dict[str, dict[str, float]], available_interface_files: set[str], ) -> None: """Sort cgroups while preserving the hierarchy.""" if not sort_names: return # Sort each level of the hierarchy rather than the flattened node list. cache = FileCache() def compare_nodes(left: CgroupNode, right: CgroupNode) -> int: for sort_name in sort_names: left_value = sort_value_for_node( left, sort_name, selectors, sampled_cpu, cache, available_interface_files, ) right_value = sort_value_for_node( right, sort_name, selectors, sampled_cpu, cache, available_interface_files, ) result = compare_sort_values(left_value, right_value, reverse_natural) if result: return result return 0 def walk(node: CgroupNode) -> None: node.children.sort(key=cmp_to_key(compare_nodes)) for child in node.children: walk(child) walk(root) ####### Table rendering ######## # Rendering is intentionally independent of cgroup semantics. It receives # final cells and headings, calculates widths, and emits deterministic text. def row_value(row: DisplayRow, header: str) -> DisplayCell: """Return the cell displayed under one table header.""" if header == "CGROUP": return DisplayCell(row.label, row.label, Align.LEFT) if header == "KEY": return DisplayCell(row.key, row.key, Align.LEFT) if header == "DEV": return DisplayCell(row.dev, row.dev, Align.LEFT) return row.cells.get(header, DisplayCell()) def is_left_aligned_header(header: str) -> bool: """Return whether an identity column is always left-aligned.""" return header in {"CGROUP", "KEY", "DEV"} def render_table( headers: Sequence[str], rows: Sequence[DisplayRow], header_lines: dict[str, tuple[str, ...]] | None = None, ) -> str: """Render rows as a deterministic plain-text table with wrapped headers.""" display_lines = { header: (header_lines or {}).get(header, (header,)) for header in headers } widths = { header: max(len(line) for line in display_lines[header]) for header in headers } for row in rows: for header in headers: widths[header] = max(widths[header], len(row_value(row, header).text)) output: list[str] = [] header_height = max(len(display_lines[header]) for header in headers) for line_index in range(header_height): parts: list[str] = [] for header in headers: lines = display_lines[header] source_index = line_index - (header_height - len(lines)) text = lines[source_index] if source_index >= 0 else "" parts.append(text.ljust(widths[header])) output.append(" ".join(parts).rstrip()) for row in rows: parts: list[str] = [] for header in headers: cell = row_value(row, header) if is_left_aligned_header(header) or cell.align == Align.LEFT: parts.append(cell.text.ljust(widths[header])) else: parts.append(cell.text.rjust(widths[header])) output.append(" ".join(parts).rstrip()) return "\n".join(output) ########################################################################### # Modification support ########################################################################### # Write requests are parsed, validated, and normalised before their target is # resolved and any kernel interface is modified. def parse_assignment( text: str, node_index: dict[str, CgroupNode] | None = None, unqualified_target: CgroupNode | None = None, ) -> Assignment: """Parse FILE=VALUE or a CGROUP_PATH/FILE=VALUE assignment.""" qualified: list[Assignment] = [] if node_index is not None: for delimiter_index, character in enumerate(text): if character != "=": continue target_and_file = text[:delimiter_index] path_rel, slash, file_name = target_and_file.rpartition("/") if not slash or not is_valid_relative_cgroup_path(path_rel): continue if not is_valid_interface_filename(file_name): continue node = node_index.get(path_rel) if node is None or not interface_exists(node, file_name): continue qualified.append( Assignment( file_name=file_name, raw_value=text[delimiter_index + 1 :], target_path_rel=path_rel, ) ) file_name, sep, raw_value = text.partition("=") unqualified_is_valid = ( bool(sep) and is_valid_interface_filename(file_name) and unqualified_target is not None and interface_exists(unqualified_target, file_name) ) if len(qualified) > 1 or (qualified and unqualified_is_valid): raise CgtreeError(f"ambiguous path-qualified --set assignment: {text}") if qualified: return qualified[0] if not sep or not file_name: raise CgtreeError(f"invalid --set value, expected FILE=VALUE: {text}") validate_interface_filename(file_name) return Assignment(file_name=file_name, raw_value=raw_value) def assignment_syntactically_targets_path(text: str, path_rel: str) -> bool: """Return whether text contains a path-qualified assignment for path_rel.""" for delimiter_index, character in enumerate(text): if character != "=": continue candidate_path, slash, file_name = text[:delimiter_index].rpartition("/") if slash and candidate_path == path_rel and is_valid_interface_filename(file_name): return True return False def normalise_int(value: str, min_value: int | None = None, max_value: int | None = None) -> str: """Validate an integer against optional bounds and return canonical text.""" try: parsed = int(value) except ValueError as exc: raise CgtreeError(f"expected integer, got: {value}") from exc if min_value is not None and parsed < min_value: raise CgtreeError(f"value {parsed} is below minimum {min_value}") if max_value is not None and parsed > max_value: raise CgtreeError(f"value {parsed} is above maximum {max_value}") return str(parsed) # Write input deliberately ignores LC_NUMERIC because the kernel expects plain # ASCII numbers. Byte parsing therefore accepts "1.5GiB" but not locale forms # such as "1,5GiB". BYTE_INPUT_RE = re.compile(r"^(?P[0-9]+(?:\.[0-9]+)?)(?P[A-Za-z]*)$") def parse_byte_input(value: str) -> int: """Parse locale-independent byte input such as 512MiB or 1.5GiB.""" match = BYTE_INPUT_RE.fullmatch(value) if not match: raise CgtreeError(f"invalid byte value: {value}") number_text = match.group("number") unit = match.group("unit").lower() multipliers = { "": 1, "b": 1, "k": 1024, "kb": 1024, "kib": 1024, "m": 1024**2, "mb": 1024**2, "mib": 1024**2, "g": 1024**3, "gb": 1024**3, "gib": 1024**3, "t": 1024**4, "tb": 1024**4, "tib": 1024**4, "p": 1024**5, "pb": 1024**5, "pib": 1024**5, "e": 1024**6, "eb": 1024**6, "eib": 1024**6, } if unit not in multipliers: raise CgtreeError(f"unknown byte suffix: {match.group('unit')}") return int(float(number_text) * multipliers[unit]) def normalise_value_by_type( value_type: str, value: str, schema: dict[str, Any] | None = None, ) -> str: """Normalise one write value according to its schema type.""" schema = schema or {} if value_type == "raw": return value if value_type == "cgroup_type": if value != "threaded": raise CgtreeError("cgroup.type accepts only threaded when written") return value if value_type == "cpuset_partition": if value not in {"member", "root", "isolated"}: raise CgtreeError("cpuset.cpus.partition accepts member, root, or isolated") return value if value_type == "bytes": return str(parse_byte_input(value)) if value_type == "bytes_or_max": return "max" if value == "max" else str(parse_byte_input(value)) if value_type in {"uclamp_percent", "uclamp_percent_or_max", "io_cost_percent"}: if value == "max" and value_type == "uclamp_percent_or_max": return value if value == "max" and value_type != "io_cost_percent": raise CgtreeError("max is only valid for a maximum utilization clamp") if not re.fullmatch(r"(?:0|[0-9]+(?:\.[0-9]+)?)", value): raise CgtreeError(f"invalid percentage value: {value}") try: parsed = Decimal(value) except InvalidOperation as exc: raise CgtreeError(f"invalid percentage value: {value}") from exc lower_bound = Decimal("1") if value_type == "io_cost_percent" else Decimal("0") upper_bound = Decimal("10000") if value_type == "io_cost_percent" else Decimal("100") if parsed < lower_bound or parsed > upper_bound: raise CgtreeError(f"percentage value outside {lower_bound}..{upper_bound}: {value}") return value if value_type == "int_or_max": return "max" if value == "max" else normalise_int(value) if value_type == "int_or_default": if value == "default": return value min_value = schema.get("min") max_value = schema.get("max") return normalise_int( value, min_value if isinstance(min_value, int) else None, max_value if isinstance(max_value, int) else None, ) if value_type == "int": min_value = schema.get("min") max_value = schema.get("max") return normalise_int( value, min_value if isinstance(min_value, int) else None, max_value if isinstance(max_value, int) else None, ) raise CgtreeError(f"unsupported value type: {value_type}") def normalise_cpu_max(value: str) -> str: """Validate and normalise a cpu.max QUOTA PERIOD value.""" tokens = value.split() if len(tokens) != 2: raise CgtreeError("cpu.max expects two fields: MAX PERIOD") quota, period = tokens if quota != "max": quota = normalise_int(quota, 1) period = normalise_int(period, 1) return f"{quota} {period}" def normalise_flat_keyed(schema: dict[str, Any], value: str) -> str: """Normalise a flat-keyed write in VALUE or KEY VALUE form.""" tokens = value.split() if len(tokens) == 1: return normalise_value_by_type(str(schema["value"]), tokens[0], schema) if len(tokens) != 2: raise CgtreeError("flat-keyed write expects VALUE or KEY VALUE") key, raw_value = tokens if key != "default" and schema.get("device_keyed", True): key = normalise_device_key(key) normalised = normalise_value_by_type(str(schema["value"]), raw_value, schema) return f"{key} {normalised}" def normalise_nested_keyed(file_name: str, schema: dict[str, Any], value: str) -> str: """Normalise a nested write in KEY FIELD=VALUE ... form.""" tokens = value.split() if len(tokens) < 2: raise CgtreeError(f"{file_name} write expects KEY FIELD=VALUE ...") key = normalise_device_key(tokens[0]) if file_name.startswith("io.") else tokens[0] fields = schema.get("fields") if not isinstance(fields, dict): raise CgtreeError(f"invalid write schema for {file_name}") output_fields: list[str] = [] for token in tokens[1:]: field_name, sep, raw_value = token.partition("=") if not sep or not field_name: raise CgtreeError(f"invalid field assignment for {file_name}: {token}") if field_name not in fields: raise CgtreeError(f"unsupported field for {file_name}: {field_name}") output_fields.append(f"{field_name}={normalise_value_by_type(str(fields[field_name]), raw_value)}") return f"{key} {' '.join(output_fields)}" def normalise_memory_reclaim(value: str) -> str: """Normalise a memory.reclaim BYTES [swappiness=VALUE] operation.""" tokens = value.split() if not tokens or len(tokens) > 2: raise CgtreeError("memory.reclaim expects BYTES [swappiness=VALUE]") amount = normalise_value_by_type("bytes", tokens[0]) if len(tokens) == 1: return amount field_name, separator, raw_value = tokens[1].partition("=") if separator != "=" or field_name != "swappiness": raise CgtreeError("memory.reclaim expects swappiness=VALUE") swappiness = "max" if raw_value == "max" else normalise_int(raw_value, 0, 200) return f"{amount} swappiness={swappiness}" def normalise_dmem_limit(value: str, value_type: str) -> str: """Normalise a DMEM REGION VALUE limit operation.""" tokens = value.split() if len(tokens) != 2: raise CgtreeError("dmem limit expects REGION VALUE") return f"{tokens[0]} {normalise_value_by_type(value_type, tokens[1])}" def writable_schema_for(file_name: str) -> dict[str, Any] | None: """Return a static or dynamic writable schema for one interface.""" schema = WRITABLE_SCHEMAS.get(file_name) if schema is not None: return schema if hugetlb_interface_kind(file_name) in {"max", "rsvd.max"}: return HUGETLB_LIMIT_WRITE_SCHEMA return None def normalise_assignment(assignment: Assignment) -> str: """Validate and normalise one --set assignment for kernel output.""" schema = writable_schema_for(assignment.file_name) if schema is None: raise CgtreeError(f"unsupported writable cgroup file: {assignment.file_name}") kind = schema.get("kind") if kind == "single": return normalise_value_by_type(str(schema["value"]), assignment.raw_value, schema) if kind == "cpu_max": return normalise_cpu_max(assignment.raw_value) if kind == "memory_reclaim": return normalise_memory_reclaim(assignment.raw_value) if kind == "dmem_limit": return normalise_dmem_limit(assignment.raw_value, str(schema["value"])) if kind == "flat_keyed": return normalise_flat_keyed(schema, assignment.raw_value) if kind == "nested_keyed": return normalise_nested_keyed(assignment.file_name, schema, assignment.raw_value) raise CgtreeError(f"unsupported write schema for {assignment.file_name}") def create_cgroup(mount_point: Path, relpath: str) -> None: """Create a cgroup path relative to the detected cgroup v2 mount.""" validate_relative_cgroup_path(relpath, "--create") path = mount_point / relpath try: path.mkdir() except FileExistsError: return except OSError as exc: raise CgtreeError(f"cannot create cgroup {relpath}: {exc}") from exc def select_write_target( node_index: dict[str, CgroupNode], selected: Sequence[CgroupNode], positional_selection: bool, created_target: CgroupNode | None, assignments: Sequence[Assignment], ) -> CgroupNode: """Resolve all modification inputs to exactly one scanned cgroup.""" target_paths: list[str] = [] if created_target is not None: target_paths.append(created_target.path_rel) if positional_selection: if not selected: raise CgtreeError("modification requires exactly one target cgroup, but none matched") if len(selected) != 1: raise CgtreeError( "modification requires exactly one target cgroup; positional MATCH matched multiple cgroups" ) target_paths.append(selected[0].path_rel) qualified_paths = { assignment.target_path_rel for assignment in assignments if assignment.target_path_rel is not None } if len(qualified_paths) > 1: raise CgtreeError("path-qualified --set assignments identify different target cgroups") if qualified_paths: target_paths.append(next(iter(qualified_paths))) if not target_paths: raise CgtreeError("modification requires a unique target selected by MATCH, --create, or path-qualified --set") unique_paths = set(target_paths) if len(unique_paths) != 1: raise CgtreeError("positional selection and path-qualified --set target different cgroups") target_path = unique_paths.pop() target = node_index.get(target_path) if target is None: raise CgtreeError(f"mutation target is no longer present: {target_path}") return target def apply_writes(target: CgroupNode, assignments: Sequence[Assignment], pids: Sequence[int]) -> None: """Prepare and apply --set and --add-pid operations to one cgroup.""" prepared: list[tuple[str, str]] = [] for assignment in assignments: validate_interface_filename(assignment.file_name) prepared.append((assignment.file_name, normalise_assignment(assignment))) required_files = [file_name for file_name, _ in prepared] if pids: required_files.append("cgroup.procs") for file_name in required_files: if not interface_exists(target, file_name): raise CgtreeError(f"cgroup interface does not exist directly beneath target: {file_name}") for file_name, normalised in prepared: write_interface_text(target, file_name, normalised) for pid in pids: write_interface_text(target, "cgroup.procs", str(pid)) ########################################################################### # Command-line interface ########################################################################### # build_parser() defines the user-facing command line. run() coordinates # discovery, selection, optional modification, sampling, sorting, display # generation, and rendering. def parse_positional_match(value: str) -> str: """Reject the removed -1/--one option while parsing MATCH.""" if value == "-1": raise argparse.ArgumentTypeError("-1/--one has been removed") return value def build_parser() -> argparse.ArgumentParser: """Build the command-line parser.""" parser = argparse.ArgumentParser( prog="cgtree", usage="cgtree [OPTIONS] [MATCH ...]", add_help=False, description=( "Inspect and modify Linux cgroup v2 hierarchies. By default, display " "first-level cgroups using the summary view." ), ) selection = parser.add_argument_group("Selection") selection.add_argument( "matches", nargs="*", metavar="MATCH", type=parse_positional_match, help=( "match cgroup basenames by substring; a value containing '/' selects " "one exact path relative to the cgroup v2 mount" ), ) selection.add_argument( "--match-mode", choices=("any", "all"), default="any", help=( "combine basename substring matchers using any or all (default: any); " "exact relative paths are matched directly" ), ) views = parser.add_argument_group("Views") views.add_argument("-m", "--memory", action="store_true", help="show memory view") views.add_argument( "-c", "--cpu", action="store_true", help="show CPU view; includes sampled CPU percentage by default", ) views.add_argument("-i", "--io", action="store_true", help="show IO view") views.add_argument("-p", "--pids", dest="show_pids", action="store_true", help="show task and process view") views.add_argument( "--comm", action="store_true", help=( "add PID and command-name rows for processes listed in cgroup.procs; " "works with built-in views and explicit selectors" ), ) views.add_argument( "-C", "--columns", metavar="SELECTORS", help="comma-separated interface selectors to display; replaces all built-in view selections", ) tree = parser.add_argument_group("Tree") tree.add_argument( "-D", "--depth", type=int, default=1, metavar="DEPTH", help="descendant depth from matched cgroups (default: 1)", ) tree.add_argument( "-R", "--recursive", action="store_true", help="show all descendants, overriding --depth", ) sorting = parser.add_argument_group("Sorting") sorting.add_argument( "-s", "--sort", nargs="?", const="", metavar="COLUMNS", help="sort by comma-separated columns (default: first displayed value column)", ) sorting.add_argument( "-ss", "--reverse-sort", nargs="?", const="", metavar="COLUMNS", help="reverse-sort by comma-separated columns (default: first displayed value column)", ) formatting = parser.add_argument_group("Formatting") formatting.add_argument( "--compact", action="store_true", help="keep multi-row interfaces on one row; incompatible with --comm", ) formatting.add_argument( "--dev-format", choices=("name", "kernel", "both"), default="name", metavar="FORMAT", help=( "format block-device keys from device-keyed interfaces as name, " "kernel major:minor, or both (default: name)" ), ) sampling = parser.add_argument_group("CPU sampling") sampling.add_argument( "--sample-period", type=float, default=1.0, metavar="SEC", help=( "CPU percentage sampling interval (default: 1.0 seconds)" ), ) sampling.add_argument( "--no-sample", action="store_true", help=( "omit sampled CPU %%" ), ) modification = parser.add_argument_group("Modification") modification.add_argument( "--set", dest="sets", action="append", default=[], metavar="ASSIGNMENT", help="write FILE=VALUE or an explicit cgroup path as CGROUP_PATH/FILE=VALUE", ) modification.add_argument( "--add-pid", dest="pids", action="append", type=int, default=[], metavar="PID", help="move PID into the selected cgroup", ) modification.add_argument( "--create", metavar="CGROUP", help="create and select a cgroup relative to the cgroup v2 mount", ) general = parser.add_argument_group("General") general.add_argument( "--list-selectors", action="store_true", help="list cgroup interface files available for use as -C selectors", ) general.add_argument("--mount", action="store_true", help="print the cgroup v2 mount point") general.add_argument("--mount-options", action="store_true", help="print cgroup v2 mount options") general.add_argument( "--version", action="version", version=f"%(prog)s {PROGRAM_VERSION}", help="show program version and exit", ) general.add_argument("-v", "--verbose", action="store_true", help="show diagnostic preamble") general.add_argument("-h", "--help", action="help", help="show this help message and exit") return parser def parse_sort_list(text: str | None) -> list[str]: """Split a comma-separated sort argument into non-empty column names.""" return [part.strip() for part in text.split(",") if part.strip()] if text else [] def default_sort_name(selectors: Sequence[SelectorSpec]) -> str: """Return the default column for --sort without an explicit argument.""" if not selectors: return "CGROUP" selector = selectors[0] if selector.full_file and selector.file_name is not None: spec = get_file_spec(selector.file_name) if spec.interface_format == InterfaceFormat.NESTED: fields = nested_display_fields(spec.name) if fields: return f"{spec.name}.{fields[0]}" return selector.name def validate_args(args: argparse.Namespace) -> None: """Validate cross-option constraints that argparse cannot express.""" if args.depth < 0: raise CgtreeError("--depth must be non-negative") if args.sample_period < 0: raise CgtreeError("--sample-period must be non-negative") if args.sort is not None and args.reverse_sort is not None: raise CgtreeError("use either --sort or --reverse-sort, not both") # --comm adds process-detail rows to any view, but those rows cannot be # represented in compact mode. if args.comm and args.compact: raise CgtreeError("--comm cannot be combined with --compact") def should_sample_cpu(args: argparse.Namespace, selectors: Sequence[SelectorSpec]) -> bool: """Return true when selected selectors require a two-point CPU sample.""" if args.no_sample: return False return any(selector.name == "cpu.percent" for selector in selectors) def run(argv: Sequence[str] | None = None) -> int: """Run cgtree and return its exit status.""" initialise_locale() parser = build_parser() args = parser.parse_args(argv) validate_args(args) mount_info = discover_cgroup2_mount() if args.mount: print(mount_info.mount_point) return 0 if args.mount_options: print(",".join(mount_info.options)) return 0 root = scan_cgroup_tree(mount_info.mount_point) node_index = build_node_index(root) available_interface_files = existing_interface_names(root) if args.list_selectors: for name in list_available_interface_files(root): print(name) return 0 if args.create: assignments = [parse_assignment(value, node_index) for value in args.sets] validate_relative_cgroup_path(args.create, "--create") if any(assignment.target_path_rel is not None for assignment in assignments) or any( assignment_syntactically_targets_path(value, args.create) for value in args.sets ): raise CgtreeError("path-qualified --set cannot be used together with --create") # Validate assignments before creating the directory so invalid input cannot # leave an empty cgroup behind. for assignment in assignments: normalise_assignment(assignment) create_cgroup(mount_info.mount_point, args.create) root = scan_cgroup_tree(mount_info.mount_point) node_index = build_node_index(root) available_interface_files = existing_interface_names(root) created_target = node_index.get(args.create) if created_target is None: raise CgtreeError(f"created cgroup was not found after scan: {args.create}") selected = [created_target] else: created_target = None selected = select_cgroups(root, node_index, args.matches, args.match_mode) if not selected: raise CgtreeError("no matching cgroups") positional_target = selected[0] if args.matches and len(selected) == 1 else None assignments = [ parse_assignment(value, node_index, unqualified_target=positional_target) for value in args.sets ] mutation_target: CgroupNode | None = None if assignments or args.pids: mutation_target = select_write_target( node_index, selected, positional_selection=bool(args.matches) and not args.create, created_target=created_target, assignments=assignments, ) assignments = [ parse_assignment(value, node_index, unqualified_target=mutation_target) for value in args.sets ] mutation_target = select_write_target( node_index, selected, positional_selection=bool(args.matches) and not args.create, created_target=created_target, assignments=assignments, ) apply_writes(mutation_target, assignments, args.pids) qualified_target_only = ( mutation_target is not None and not args.matches and not args.create and any(assignment.target_path_rel is not None for assignment in assignments) ) if qualified_target_only: assert mutation_target is not None selected = [mutation_target] selectors, pretty_headers = resolve_selectors(args, available_interface_files) depth = None if args.recursive else args.depth explicit_selection = bool(args.matches or args.create or qualified_target_only) include_context = bool((args.matches and not args.create) or qualified_target_only) display_roots = selected if explicit_selection else root.children display_depth = depth if explicit_selection or args.recursive else 0 # Sample after selection but before sorting and display because CPU percentage # may itself be used as a sort column. pre_sort_nodes, pre_sort_data = visible_nodes(root, display_roots, display_depth, include_context) sampled_cpu: dict[str, dict[str, float]] = {} if should_sample_cpu(args, selectors): sample_nodes = [node for node in pre_sort_nodes if node.path_rel in pre_sort_data] sampled_cpu = sample_cpu_percent(sample_nodes, args.sample_period) sort_option = args.sort if args.sort is not None else args.reverse_sort sort_names = parse_sort_list(sort_option) if sort_option is not None and not sort_names: sort_names = [default_sort_name(selectors)] sort_tree( root, sort_names, args.reverse_sort is not None, selectors, sampled_cpu, available_interface_files, ) # Rebuild the visible-node order after sibling sorting. rows_nodes, data_relpaths = visible_nodes(root, display_roots, display_depth, include_context) options = DisplayOptions( compact=args.compact, dev_format=args.dev_format, pretty_headers=pretty_headers, ) headers, header_lines, rows = build_display_rows( root, rows_nodes, data_relpaths, selectors, options, sampled_cpu, show_comm=args.comm, ) if args.verbose: print(f"cgtree {PROGRAM_VERSION}") print(f"mount: {mount_info.mount_point}") print(f"matches: {', '.join(args.matches) if args.matches else '(none)'}") if should_sample_cpu(args, selectors): print(f"sample period: {args.sample_period:g}s") print(render_table(headers, rows, header_lines)) return 0 def main(argv: Sequence[str] | None = None) -> int: """Run the command-line entry point with user-facing error handling.""" try: return run(argv) except CgtreeError as exc: print(f"cgtree: {exc}", file=sys.stderr) return 1 except BrokenPipeError: return 1 if __name__ == "__main__": raise SystemExit(main())