From 87b2adf845906880aae029443ef5ac2ff0762080 Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Thu, 6 Aug 2026 22:51:21 +0000 Subject: [PATCH 1/2] respect container memory and CPU limits from cgroups --- docs/guides/scaling_crawlers.mdx | 6 + src/crawlee/_utils/cgroup.py | 434 +++++++++++++++++ src/crawlee/_utils/system.py | 170 ++++++- tests/unit/_utils/test_cgroup.py | 784 +++++++++++++++++++++++++++++++ 4 files changed, 1385 insertions(+), 9 deletions(-) create mode 100644 src/crawlee/_utils/cgroup.py create mode 100644 tests/unit/_utils/test_cgroup.py diff --git a/docs/guides/scaling_crawlers.mdx b/docs/guides/scaling_crawlers.mdx index 152d852e60..a720c57047 100644 --- a/docs/guides/scaling_crawlers.mdx +++ b/docs/guides/scaling_crawlers.mdx @@ -47,3 +47,9 @@ The `desired_concurrency` option in the ## Autoscaled pool The `AutoscaledPool` manages a pool of asynchronous, resource-intensive tasks that run in parallel. It automatically starts new tasks only when there is enough free CPU and memory. To monitor system resources, it leverages the `Snapshotter` and `SystemStatus` classes. If any task raises an exception, the error is propagated, and the pool is stopped. Every crawler uses an `AutoscaledPool` under the hood. + +## Running under a resource limit + +A crawler often gets less than the host machine has. A Docker container, a Kubernetes pod and a systemd slice each carry a limit of their own. Crawlee reads that limit from the cgroup and scales against it, so it doesn't have to be told about it. The memory budget comes from the memory limit, the CPU load is measured against the cores the process may use, and the tightest limit wins when several of them apply. Outside Linux, and without a limit, Crawlee falls back to the resources of the host machine. + +The budget is `available_memory_ratio` of the limit, 25% by default, and the crawler throttles once it uses `max_used_memory_ratio` of that budget, 90% by default. A container limited to 2 GB therefore throttles at around 460 MB. Both options live in the `Configuration`, together with `memory_mbytes` for sizing the budget in absolute terms. diff --git a/src/crawlee/_utils/cgroup.py b/src/crawlee/_utils/cgroup.py new file mode 100644 index 0000000000..f094b1bbfc --- /dev/null +++ b/src/crawlee/_utils/cgroup.py @@ -0,0 +1,434 @@ +from __future__ import annotations + +from dataclasses import dataclass +from functools import lru_cache +from logging import WARNING, getLogger +from pathlib import Path, PurePosixPath + +from crawlee._utils.log import LoggerOnce + +logger = getLogger(__name__) +logger_once = LoggerOnce(logger) + +_PROC_SELF_CGROUP = Path('/proc/self/cgroup') +"""Lists the cgroup this process belongs to, one line per mounted hierarchy.""" + +_PROC_SELF_MOUNTINFO = Path('/proc/self/mountinfo') +"""Lists the mounted filesystems. Read to locate the hierarchies instead of assuming `/sys/fs/cgroup`.""" + +_MICROSECONDS_PER_SECOND = 1_000_000 +_NANOSECONDS_PER_SECOND = 1_000_000_000 + +_V1_CONTROLLER_NAMES = frozenset({'memory', 'cpu', 'cpuacct', 'cpuset'}) +"""The controllers worth recording. A cgroup v1 mount lists them among options that name no controller at all.""" + + +@dataclass(frozen=True) +class _Hierarchy: + """A mounted cgroup hierarchy, together with the cgroup this process belongs to in it.""" + + point: Path + """The directory the hierarchy is mounted at.""" + + root: str + """The subtree of the hierarchy the mount exposes, spelled the same way as `own_path`.""" + + own_path: str + """The cgroup this process belongs to, as `/proc/self/cgroup` spells it.""" + + +@dataclass(frozen=True) +class _Controller: + """A located controller: the interface it provides and the directories to read its control files from.""" + + is_v2: bool + """Whether the controller provides the cgroup v2 interface, which spells its control files differently.""" + + dirs: tuple[Path, ...] + """The cgroup of this process first, then each of its ancestors up to the top of the hierarchy. + + A limit set on an ancestor caps everything below it. Under Kubernetes the container, the pod and the QoS class + each get a level of their own, and the limit that matters may sit on any of them. + """ + + +@dataclass(frozen=True) +class _Controllers: + """The controllers that carry resource metrics, as located for this process.""" + + memory: _Controller | None + """The controller carrying the memory limit and the memory charged against it.""" + + cpu_quota: _Controller | None + """The controller carrying the CPU bandwidth quota.""" + + cpu_usage: _Controller | None + """The controller carrying the consumed CPU time. Under cgroup v1 that is `cpuacct`, a controller of its own.""" + + cpu_set: _Controller | None + """The controller carrying the set of cores the cgroup may run on.""" + + +@dataclass(frozen=True) +class MemoryLimit: + """A cgroup memory limit, together with the memory usage charged against it.""" + + limit: int + """The tightest limit applying to this process, in bytes.""" + + working_set: int + """The memory charged against the limit, in bytes, excluding reclaimable file cache. + + Several cgroups can hold a limit, and the one closest to running out is not always the one holding the tightest + limit. This is the highest utilization among them, expressed against `limit` so that the two stay comparable. + """ + + +def get_memory_limit() -> MemoryLimit | None: + """Get the tightest memory limit applying to this process, with the usage measured against it. + + A usage read at one level next to a limit read at another compares two different scopes, and the ratio between + them then says nothing about how close an out-of-memory kill is. Both numbers are therefore brought to the same + scale before being returned. + + Returns: + The limit and the working set, or `None` when no cgroup limit applies to this process or the control files + cannot be read. + """ + controller = _get_controllers().memory + if controller is None: + return None + + # Only the hard limit counts. A cgroup can sit above `memory.high` indefinitely, because it throttles reclaim + # rather than triggering an out-of-memory kill. + file_name = 'memory.max' if controller.is_v2 else 'memory.limit_in_bytes' + + # Exceeding the tightest limit gets the process killed whichever level holds it, so that one is the ceiling a + # memory budget has to fit under. Which level runs out first is a separate question, because a limit further up + # covers the sibling cgroups too. The utilization is therefore taken from the level closest to its own limit, + # wherever along the chain that sits. + ceiling: int | None = None + used_ratios = [] + + for directory in controller.dirs: + limit = _read_int(directory / file_name) + if limit is None or limit <= 0: + continue + + # A level counts towards the ceiling on the strength of its limit alone. Its usage is a separate reading that + # can be missing, and dropping the limit along with it would raise the ceiling above what the kernel enforces. + ceiling = limit if ceiling is None else min(ceiling, limit) + + working_set = _read_working_set(controller, directory) + if working_set is not None: + used_ratios.append(working_set / limit) + + if ceiling is None: + return None + + if not used_ratios: + logger_once.log( + 'Found a cgroup memory limit but no usage metric to pair it with, so the limit is ignored - the ' + 'autoscaler may scale past the limit of this container.', + key='cgroup_memory_usage_unavailable', + level=WARNING, + ) + return None + + # A cgroup can sit above its limit while the kernel reclaims, which is not a usage the caller can act on. + used_ratio = min(max(used_ratios), 1.0) + + return MemoryLimit(limit=ceiling, working_set=round(used_ratio * ceiling)) + + +def get_cpu_quota() -> float | None: + """Get the number of CPU cores the bandwidth quota allows this process to use. + + Returns: + The number of cores, which can be fractional, or `None` when no quota applies to this process or the control + files cannot be read. + """ + controller = _get_controllers().cpu_quota + if controller is None: + return None + + read_quota = _read_cpu_quota_v2 if controller.is_v2 else _read_cpu_quota_v1 + + # Levels that hold no readable quota drop out, so an unlimited ancestor does not hide a quota set below it. + quotas = [quota for directory in controller.dirs if (quota := read_quota(directory)) is not None] + + return min(quotas) if quotas else None + + +def get_cpu_set_size() -> int | None: + """Get the number of CPU cores the cgroup of this process is allowed to run on. + + Reading the cgroup rather than the affinity of the process keeps this in the same scope as the CPU time it gets + paired with. A `taskset` narrows one process without narrowing the cgroup around it. + + Returns: + The number of cores, or `None` when no CPU set applies to this process or the control file cannot be read. + """ + controller = _get_controllers().cpu_set + if controller is None: + return None + + # The effective set already accounts for the ancestors, because a cgroup is never given cores its parent lacks. + file_name = 'cpuset.cpus.effective' if controller.is_v2 else 'cpuset.cpus' + + try: + cpu_list = (controller.dirs[0] / file_name).read_text().strip() + except OSError: + return None + + # Under cgroup v1 an empty set means the cgroup inherits the cores of its parent. + return _count_cpu_list(cpu_list) if cpu_list else None + + +def get_cpu_usage() -> float | None: + """Get the CPU time the cgroup of this process has consumed since the cgroup was created, in seconds. + + Unlike the memory metrics, this reads the own cgroup rather than the level the quota sits on. A quota set on an + ancestor alone is not how container runtimes spell a CPU limit, so the two levels coincide in practice. + + Returns: + The cumulative CPU time, or `None` when it cannot be read. + """ + controller = _get_controllers().cpu_usage + if controller is None: + return None + + own_dir = controller.dirs[0] + + if controller.is_v2: + microseconds = _read_stat_value(own_dir / 'cpu.stat', 'usage_usec') + return microseconds / _MICROSECONDS_PER_SECOND if microseconds is not None else None + + nanoseconds = _read_int(own_dir / 'cpuacct.usage') + return nanoseconds / _NANOSECONDS_PER_SECOND if nanoseconds is not None else None + + +def _read_working_set(controller: _Controller, directory: Path) -> int | None: + """Read the memory charged to one cgroup, in bytes, excluding reclaimable file cache. + + The raw usage a cgroup reports counts the page cache, which the kernel drops on demand and which therefore does + not predict an out-of-memory kill. Subtracting the inactive file cache gives the working set, the same figure + `docker stats`, `kubectl top` and cAdvisor report. + """ + current = _read_int(directory / ('memory.current' if controller.is_v2 else 'memory.usage_in_bytes')) + if current is None: + return None + + # Falling back to the raw usage would turn this into "usage including reclaimable cache", which is the confusion + # the subtraction exists to prevent. + inactive_file_key = 'inactive_file' if controller.is_v2 else 'total_inactive_file' + inactive_file = _read_stat_value(directory / 'memory.stat', inactive_file_key) + if inactive_file is None: + return None + + return max(current - inactive_file, 0) + + +@lru_cache(maxsize=1) +def _get_controllers() -> _Controllers: + """Locate the control files that carry the resource metrics of this process. + + Locating them walks `/proc`, which costs orders of magnitude more than reading a single control file, and the + result stays valid for the lifetime of the process. The control files themselves are read again on every sample. + """ + try: + unified, v1 = _read_hierarchies() + except OSError: + # Either this is not Linux, or `/proc` is not mounted. No cgroup metrics exist to be read in both cases. + return _Controllers(memory=None, cpu_quota=None, cpu_usage=None, cpu_set=None) + + return _Controllers( + memory=_locate_controller(unified, v1, 'memory', v2_probe='memory.current', v1_probe='memory.usage_in_bytes'), + cpu_quota=_locate_controller(unified, v1, 'cpu', v2_probe='cpu.max', v1_probe='cpu.cfs_quota_us'), + # Under cgroup v1 the quota and the accounting belong to two controllers that can be mounted separately. + cpu_usage=_locate_controller(unified, v1, 'cpuacct', v2_probe='cpu.stat', v1_probe='cpuacct.usage'), + cpu_set=_locate_controller(unified, v1, 'cpuset', v2_probe='cpuset.cpus.effective', v1_probe='cpuset.cpus'), + ) + + +def _locate_controller( + unified: _Hierarchy | None, + v1: dict[str, _Hierarchy], + v1_name: str, + *, + v2_probe: str, + v1_probe: str, +) -> _Controller | None: + """Locate the directories holding the files of one controller, preferring the cgroup v2 unified hierarchy. + + A system can mount both interfaces at once with only some of the controllers enabled on the unified hierarchy, so + a candidate counts only where the probe file it is supposed to provide exists. + """ + for hierarchy, probe_file, is_v2 in ((unified, v2_probe, True), (v1.get(v1_name), v1_probe, False)): + if hierarchy is None: + continue + + dirs = _trim_to_controller(_candidate_dirs(hierarchy), probe_file) + if dirs: + return _Controller(is_v2=is_v2, dirs=dirs) + + return None + + +def _trim_to_controller(dirs: tuple[Path, ...], probe_file: str) -> tuple[Path, ...]: + """Drop the levels below the closest one that carries the controller.""" + # A cgroup gets a controller's files only once its parent enables that controller for its children, so the levels + # closest to the process can be missing them while the levels above still declare the limit that applies. + for index, directory in enumerate(dirs): + if (directory / probe_file).exists(): + return dirs[index:] + + return () + + +def _candidate_dirs(hierarchy: _Hierarchy) -> tuple[Path, ...]: + """List the directories a controller's files can be read from, the cgroup of this process first.""" + path = PurePosixPath(hierarchy.own_path) + + # A mount can expose just a subtree of a hierarchy, and then the paths in `/proc/self/cgroup` carry the mount root + # as a prefix that has to come off. Container runtimes instead give the container a cgroup namespace of its own. + if hierarchy.root != '/': + try: + path = PurePosixPath('/') / path.relative_to(hierarchy.root) + except ValueError: + # The mount does not cover the cgroup of this process, so only the top of the mount is worth reading. + path = PurePosixPath('/') + + parts = path.parts[1:] if path.is_absolute() else path.parts + own_dir = hierarchy.point.joinpath(*parts) + + # Walking up stops at the mount point, because nothing above it belongs to the hierarchy. + return (own_dir, *own_dir.parents[: len(parts)]) + + +def _read_hierarchies() -> tuple[_Hierarchy | None, dict[str, _Hierarchy]]: + """Locate the mounted cgroup hierarchies and the cgroup this process belongs to in each of them. + + Returns: + The cgroup v2 unified hierarchy, and the cgroup v1 hierarchies keyed by the controller each one carries. + + Raises: + OSError: If `/proc/self/mountinfo` or `/proc/self/cgroup` cannot be read. + """ + unified_path, controller_paths = _read_own_paths() + + unified: _Hierarchy | None = None + controllers: dict[str, _Hierarchy] = {} + + for line in _PROC_SELF_MOUNTINFO.read_text().splitlines(): + # A variable number of optional fields sits between the mount point and the ` - ` separator, so the line has + # to be split on the separator first. + before, separator, after = line.partition(' - ') + if not separator: + continue + + try: + _mount_id, _parent_id, _device, mount_root, mount_point, *_ = before.split(' ') + filesystem, _source, super_options, *_ = after.split(' ') + except ValueError: + continue + + if filesystem == 'cgroup2' and unified_path is not None: + # The same hierarchy can be bind-mounted a second time, for instance by an agent that watches the host + # from inside a container. Mounts are listed in the order they were made, so the first one is ours. + unified = unified or _Hierarchy(point=Path(mount_point), root=mount_root, own_path=unified_path) + elif filesystem == 'cgroup': + # A cgroup v1 mount names the controllers it carries among its super options, e.g. `rw,cpu,cpuacct`. + for option in super_options.split(','): + own_path = controller_paths.get(option) + if option in _V1_CONTROLLER_NAMES and own_path is not None: + hierarchy = _Hierarchy(point=Path(mount_point), root=mount_root, own_path=own_path) + controllers.setdefault(option, hierarchy) + + return unified, controllers + + +def _read_own_paths() -> tuple[str | None, dict[str, str]]: + """Read the cgroup this process belongs to in the unified hierarchy and in each cgroup v1 one. + + Raises: + OSError: If `/proc/self/cgroup` cannot be read. + """ + unified: str | None = None + controllers: dict[str, str] = {} + + for line in _PROC_SELF_CGROUP.read_text().splitlines(): + try: + _hierarchy_id, controller_list, cgroup_path = line.split(':', 2) + except ValueError: + continue + + # The unified hierarchy is the entry with no controllers listed, spelled `0::`. + if not controller_list: + unified = cgroup_path + continue + + for controller in controller_list.split(','): + # A cgroup v1 hierarchy mounted without a controller carries a name instead, e.g. `name=systemd`. + controllers[controller.removeprefix('name=')] = cgroup_path + + return unified, controllers + + +def _count_cpu_list(cpu_list: str) -> int | None: + """Count the CPUs a control file lists as a mix of ranges and single numbers, e.g. `0-3,8`.""" + count = 0 + + try: + for part in cpu_list.split(','): + first, separator, last = part.partition('-') + count += int(last) - int(first) + 1 if separator else 1 + except ValueError: + return None + + return count + + +def _read_cpu_quota_v2(directory: Path) -> float | None: + """Read the number of cores allowed by the quota and the period a cgroup v2 `cpu.max` file holds.""" + try: + quota, period = (directory / 'cpu.max').read_text().split() + # An unlimited cgroup spells the quota as `max`, which is not an integer. + return int(quota) / int(period) + except (OSError, ValueError, ZeroDivisionError): + return None + + +def _read_cpu_quota_v1(directory: Path) -> float | None: + """Read the number of cores the cgroup v1 quota and period files allow.""" + quota = _read_int(directory / 'cpu.cfs_quota_us') + period = _read_int(directory / 'cpu.cfs_period_us') + + # An unlimited cgroup sets the quota to a negative value. + if quota is None or period is None or quota <= 0 or period <= 0: + return None + + return quota / period + + +def _read_int(path: Path) -> int | None: + """Read a control file holding a single integer, or `None` if it holds anything else.""" + try: + return int(path.read_text().strip()) + except (OSError, ValueError): + # cgroup v2 spells an absent limit as `max`, which is not an integer. + return None + + +def _read_stat_value(path: Path, key: str) -> int | None: + """Read one entry of a control file holding a flat table of ` ` lines.""" + try: + with path.open() as file: + for line in file: + entry_key, _separator, value = line.partition(' ') + if entry_key == key: + return int(value) + except (OSError, ValueError): + return None + + return None diff --git a/src/crawlee/_utils/system.py b/src/crawlee/_utils/system.py index 45d0483679..1f06dd4551 100644 --- a/src/crawlee/_utils/system.py +++ b/src/crawlee/_utils/system.py @@ -2,6 +2,8 @@ import os import sys +import time +from dataclasses import dataclass from datetime import datetime, timezone from logging import WARNING, getLogger from typing import TYPE_CHECKING, Annotated @@ -9,6 +11,7 @@ import psutil from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, PlainValidator +from crawlee._utils import cgroup from crawlee._utils.byte_size import ByteSize from crawlee._utils.log import LoggerOnce @@ -19,6 +22,9 @@ # psutil re-raises `FileNotFoundError` as is when a `/proc` entry is missing for a process that is still alive. _METRIC_ERRORS = (psutil.Error, OSError) +_CPU_SAMPLE_INTERVAL_SECS = 0.1 +"""How long a CPU measurement waits when it has no earlier reading to compare against.""" + class _PssAvailability: """Process-wide latch for whether the PSS memory metric exists on this system at all. @@ -185,7 +191,11 @@ class MemoryInfo(MemoryUsageInfo): total_size: Annotated[ ByteSize, PlainValidator(ByteSize.validate), PlainSerializer(lambda size: size.bytes), Field(alias='totalSize') ] - """Total memory available in the system.""" + """Total memory available to this process. + + In a container with a memory limit this is the limit, not the memory of the host machine, so that a budget derived + from it cannot exceed what the container is allowed to use. + """ system_wide_used_size: Annotated[ ByteSize, @@ -193,25 +203,138 @@ class MemoryInfo(MemoryUsageInfo): PlainSerializer(lambda size: size.bytes), Field(alias='systemWideUsedSize'), ] - """Total memory used by all processes system-wide (including non-crawlee processes).""" + """Total memory used within the scope `total_size` covers, including memory used by non-crawlee processes. + + In a container with a memory limit this is the working set charged against that limit, the same kind of figure + `docker stats` and `kubectl top` report. Elsewhere it is the memory used across the host machine. + """ + + +@dataclass(frozen=True) +class _CpuReading: + """A cumulative CPU time reading of a cgroup, with the moment it was taken.""" + + used_seconds: float + """The CPU time the cgroup has consumed since it was created, in seconds.""" + + taken_at: float + """The value of the monotonic clock at the moment the reading was taken.""" + + +class _CgroupCpu: + """Process-wide latch holding the previous cgroup CPU reading. + + A cgroup exposes consumed CPU time as a counter that only grows, so a usage ratio takes two readings taken at + different moments. `psutil.cpu_percent` keeps equivalent state of its own internally. Only the event manager + samples the CPU, one call at a time, so the latch needs no synchronization of its own. + """ + + previous: _CpuReading | None = None + + +def _get_cpu_set_size() -> float | None: + """Get the number of cores the cgroup of this process is pinned to, or `None` when that is not a restriction.""" + host_cores = psutil.cpu_count() + cpu_set_cores = cgroup.get_cpu_set_size() + + if host_cores is None or cpu_set_cores is None or cpu_set_cores >= host_cores: + return None + + return float(cpu_set_cores) + + +def _get_allowed_cpu_cores() -> float | None: + """Get the number of CPU cores this process may use, or `None` when nothing restricts it. + + A bandwidth quota and a CPU set restrict it independently, and the tighter one wins. Both are read from the cgroup + of this process, so they cover the same scope as the CPU time they get paired with. + """ + limits = [limit for limit in (cgroup.get_cpu_quota(), _get_cpu_set_size()) if limit is not None] + + return min(limits) if limits else None + + +def _log_cpu_usage_unavailable() -> None: + """Report that a CPU limit was found but cannot be measured against, so the whole machine is measured instead.""" + logger_once.log( + 'Found a cgroup CPU limit but no usage metric to pair it with, so the load of the whole machine is measured ' + 'instead - the autoscaler may scale past the CPU this container is allowed to use.', + key='cgroup_cpu_usage_unavailable', + level=WARNING, + ) + + +def _take_cpu_reading() -> _CpuReading | None: + """Read the cumulative CPU time of the cgroup of this process, with the moment the reading was taken.""" + used_seconds = cgroup.get_cpu_usage() + + return _CpuReading(used_seconds=used_seconds, taken_at=time.monotonic()) if used_seconds is not None else None + + +def _get_cgroup_cpu_used_ratio() -> float | None: + """Get the CPU usage of the cgroup of this process relative to the cores it is allowed to use. + + Returns: + The ratio, or `None` when nothing restricts the CPU of this process or the cgroup metrics cannot be read. + """ + allowed_cores = _get_allowed_cpu_cores() + if allowed_cores is None: + return None + + previous = _CgroupCpu.previous + if previous is None: + # A counter that only grows takes two readings to turn into a rate. Waiting for the second one costs the same + # as the interval `psutil` spends on its own first call, and keeps every sample in the scope of the cgroup. + previous = _take_cpu_reading() + if previous is None: + _log_cpu_usage_unavailable() + return None + + # Kept even if the reading below fails, because a counter that only grows can be compared against any earlier + # reading. Without this, every later sample would wait out the interval again. + _CgroupCpu.previous = previous + time.sleep(_CPU_SAMPLE_INTERVAL_SECS) + + reading = _take_cpu_reading() + if reading is None: + _log_cpu_usage_unavailable() + return None + + _CgroupCpu.previous = reading + + elapsed_seconds = reading.taken_at - previous.taken_at + if elapsed_seconds <= 0: + return None + + used_ratio = (reading.used_seconds - previous.used_seconds) / (elapsed_seconds * allowed_cores) + + # Moving the process to another cgroup restarts the counter, which makes the difference negative. + return min(max(used_ratio, 0.0), 1.0) def get_cpu_info() -> CpuInfo: """Retrieve the current CPU usage. - It utilizes the `psutil` library. Function `psutil.cpu_percent()` returns a float representing the current - system-wide CPU utilization as a percentage. + When a container restricts the CPU of this process, the usage is measured against the cores that container may + use, read from its cgroup. Without such a restriction the process competes for the whole machine, so the + system-wide utilization reported by `psutil.cpu_percent()` is used instead. """ logger.debug('Calling get_cpu_info()...') - cpu_percent = psutil.cpu_percent(interval=0.1) - return CpuInfo(used_ratio=cpu_percent / 100) + + used_ratio = _get_cgroup_cpu_used_ratio() + + if used_ratio is None: + used_ratio = psutil.cpu_percent(interval=_CPU_SAMPLE_INTERVAL_SECS) / 100 + + return CpuInfo(used_ratio=used_ratio) def get_memory_info() -> MemoryInfo: """Retrieve the current memory usage of the process and its children. It utilizes the `psutil` library. The reported `current_size` is best-effort - processes that cannot be inspected - are left out of the sum, and PSS may be substituted by RSS for some or all of the processes. + are left out of the sum, and PSS may be substituted by RSS for some or all of the processes. The system-wide + figures come from the cgroup of this process whenever it limits how much memory the process may use. """ logger.debug('Calling get_memory_info()...') current_process = psutil.Process(os.getpid()) @@ -237,9 +360,38 @@ def get_memory_info() -> MemoryInfo: current_size_bytes += _get_child_used_memory(child) vm = psutil.virtual_memory() + total_size_bytes, system_wide_used_size_bytes = _get_system_wide_memory( + host_total_bytes=vm.total, + host_used_bytes=vm.total - vm.available, + ) return MemoryInfo( - total_size=ByteSize(vm.total), + total_size=ByteSize(total_size_bytes), current_size=ByteSize(current_size_bytes), - system_wide_used_size=ByteSize(vm.total - vm.available), + system_wide_used_size=ByteSize(system_wide_used_size_bytes), ) + + +def _get_system_wide_memory(*, host_total_bytes: int, host_used_bytes: int) -> tuple[int, int]: + """Narrow the system-wide memory metrics down to the cgroup this process runs in. + + `/proc/meminfo`, which `psutil` reads, is not namespaced, so in a memory-limited container it reports the memory + of the host instead of the limit the process actually has. Left uncorrected, the autoscaler derives its budget + from the host total and can keep scaling until the container is killed instead of throttling. + + Args: + host_total_bytes: Total memory of the host. + host_used_bytes: Memory used across the host. + + Returns: + The total and the used memory to report. + """ + memory_limit = cgroup.get_memory_limit() + + # A cgroup without a memory limit reports one at least as large as the memory of the host. cgroup v1 spells it as + # a sentinel close to the largest signed 64-bit integer, and runtimes differ on its exact value, so the host total + # is the reliable thing to compare against. + if memory_limit is None or memory_limit.limit >= host_total_bytes: + return host_total_bytes, host_used_bytes + + return memory_limit.limit, memory_limit.working_set diff --git a/tests/unit/_utils/test_cgroup.py b/tests/unit/_utils/test_cgroup.py new file mode 100644 index 0000000000..8d0dc9a6c8 --- /dev/null +++ b/tests/unit/_utils/test_cgroup.py @@ -0,0 +1,784 @@ +from __future__ import annotations + +import logging +from itertools import count +from types import SimpleNamespace +from typing import TYPE_CHECKING + +import psutil +import pytest + +from crawlee._utils import cgroup, system +from crawlee._utils.byte_size import ByteSize +from crawlee._utils.log import LoggerOnce +from crawlee._utils.system import get_cpu_info, get_memory_info + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + from pathlib import Path + +V2_MOUNTINFO = '25 30 0:22 / {root} rw,nosuid,nodev,noexec,relatime shared:4 - cgroup2 cgroup2 rw,nsdelegate' +"""A single unified hierarchy exposed from its top, which is what a container runtime sets up.""" + +V1_MOUNTINFO = ( + '29 25 0:25 / {root}/systemd rw,nosuid shared:9 - cgroup cgroup rw,name=systemd\n' + '30 25 0:26 / {root}/memory rw,nosuid shared:14 - cgroup cgroup rw,memory\n' + '31 25 0:27 / {root}/cpu,cpuacct rw,nosuid shared:15 - cgroup cgroup rw,cpu,cpuacct\n' + '32 25 0:28 / {root}/cpuset rw,nosuid shared:16 - cgroup cgroup rw,cpuset' +) +"""One hierarchy per controller, with the CPU accounting sharing a mount with the CPU bandwidth controller.""" + +V2_SELF_CGROUP = '0::{path}\n' +"""The unified hierarchy is the entry with no controllers listed.""" + +V1_SELF_CGROUP = '4:cpuset:{path}\n3:cpu,cpuacct:{path}\n2:memory:{path}\n1:name=systemd:{path}\n' +"""One entry per cgroup v1 hierarchy, including the named one that carries no controller.""" + +HOST_TOTAL_BYTES = 8 * 1024**3 +HOST_AVAILABLE_BYTES = 3 * 1024**3 + + +@pytest.fixture(autouse=True) +def _isolated_module_state(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + """Reset the process-wide state of both modules, so that nothing leaks between tests.""" + monkeypatch.setattr(system, 'logger_once', LoggerOnce(system.logger)) + monkeypatch.setattr(cgroup, 'logger_once', LoggerOnce(cgroup.logger)) + monkeypatch.setattr(system._CgroupCpu, 'previous', None) + cgroup._get_controllers.cache_clear() + yield + cgroup._get_controllers.cache_clear() + + +@pytest.fixture +def fake_cgroup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Callable[..., Path]: + """Return a builder that lays out a fake cgroup filesystem, points the module at it and returns its root.""" + + def build(*, mountinfo: str, self_cgroup: str, files: dict[str, str]) -> Path: + root = tmp_path / 'cgroup' + root.mkdir(parents=True, exist_ok=True) + + for relative_path, content in files.items(): + path = root / relative_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content) + + mountinfo_path = tmp_path / 'mountinfo' + mountinfo_path.write_text(mountinfo.format(root=root)) + self_cgroup_path = tmp_path / 'self_cgroup' + self_cgroup_path.write_text(self_cgroup) + + monkeypatch.setattr(cgroup, '_PROC_SELF_MOUNTINFO', mountinfo_path) + monkeypatch.setattr(cgroup, '_PROC_SELF_CGROUP', self_cgroup_path) + + return root + + return build + + +@pytest.fixture +def _no_cgroup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Point the module at `/proc` files that do not exist, as on a system without cgroups.""" + monkeypatch.setattr(cgroup, '_PROC_SELF_MOUNTINFO', tmp_path / 'missing') + monkeypatch.setattr(cgroup, '_PROC_SELF_CGROUP', tmp_path / 'missing') + + +@pytest.fixture +def _fixed_host_memory(monkeypatch: pytest.MonkeyPatch) -> None: + """Pin the host memory `psutil` reports, so the expected values do not move with the machine running the tests.""" + monkeypatch.setattr( + psutil, + 'virtual_memory', + lambda: SimpleNamespace(total=HOST_TOTAL_BYTES, available=HOST_AVAILABLE_BYTES), + ) + + +@pytest.fixture +def _one_second_per_sample(monkeypatch: pytest.MonkeyPatch) -> None: + """Advance the clock the module reads by one second per reading, so a usage rate comes out predictable.""" + clock = count(start=100.0, step=1.0) + monkeypatch.setattr(system, 'time', SimpleNamespace(monotonic=lambda: next(clock), sleep=lambda _seconds: None)) + + +def test_read_hierarchies_v2(fake_cgroup: Callable[..., Path]) -> None: + """Finds the unified hierarchy and the cgroup this process belongs to in it.""" + root = fake_cgroup(mountinfo=V2_MOUNTINFO, self_cgroup=V2_SELF_CGROUP.format(path='/init.scope'), files={}) + + unified, controllers = cgroup._read_hierarchies() + + assert unified is not None + assert unified.point == root + assert unified.root == '/' + assert unified.own_path == '/init.scope' + assert controllers == {} + + +def test_read_hierarchies_v1(fake_cgroup: Callable[..., Path]) -> None: + """Finds every controller of a cgroup v1 mount that carries more than one.""" + root = fake_cgroup(mountinfo=V1_MOUNTINFO, self_cgroup=V1_SELF_CGROUP.format(path='/docker/abc'), files={}) + + unified, controllers = cgroup._read_hierarchies() + + assert unified is None + assert controllers['memory'].point == root / 'memory' + assert controllers['memory'].own_path == '/docker/abc' + assert controllers['cpuset'].point == root / 'cpuset' + # Otherwise the CPU metrics get split across versions. + assert controllers['cpu'].point == root / 'cpu,cpuacct' + assert controllers['cpuacct'].point == root / 'cpu,cpuacct' + + +def test_read_hierarchies_bad_lines(fake_cgroup: Callable[..., Path]) -> None: + """Skips the lines it cannot parse, which a table of every filesystem on the machine is full of.""" + mountinfo = f'not a mount line\n24 30 0:21 / /sys rw - sysfs sysfs rw\n{V2_MOUNTINFO}\n25 30 0:22 /' + root = fake_cgroup(mountinfo=mountinfo, self_cgroup=V2_SELF_CGROUP.format(path='/'), files={}) + + unified, _controllers = cgroup._read_hierarchies() + + assert unified is not None + assert unified.point == root + + +@pytest.mark.parametrize( + ('self_cgroup', 'expected_unified', 'expected_controllers'), + [ + pytest.param('0::/init.scope\n', '/init.scope', {}, id='unified only'), + pytest.param( + '2:memory:/docker/abc\n1:name=systemd:/docker/abc\n', + None, + {'memory': '/docker/abc', 'systemd': '/docker/abc'}, + id='v1 only, with a named hierarchy', + ), + pytest.param( + '2:memory:/system.slice\n0::/user.slice\n', + '/user.slice', + {'memory': '/system.slice'}, + id='both interfaces at once', + ), + pytest.param('nonsense\n0::/\n', '/', {}, id='unparsable line skipped'), + ], +) +def test_read_own_paths( + fake_cgroup: Callable[..., Path], + self_cgroup: str, + expected_unified: str | None, + expected_controllers: dict[str, str], +) -> None: + """Reads the cgroup this process belongs to in each hierarchy.""" + fake_cgroup(mountinfo=V2_MOUNTINFO, self_cgroup=self_cgroup, files={}) + + unified, controllers = cgroup._read_own_paths() + + assert unified == expected_unified + assert controllers == expected_controllers + + +@pytest.mark.parametrize( + ('mount_root', 'cgroup_path', 'expected'), + [ + pytest.param('/', '/', [''], id='own cgroup at the top of the mount'), + pytest.param( + '/', + '/kubepods/pod/container', + ['kubepods/pod/container', 'kubepods/pod', 'kubepods', ''], + id='nested', + ), + pytest.param('/docker/abc', '/docker/abc/nested', ['nested', ''], id='mount root stripped'), + pytest.param('/docker/abc', '/system.slice', [''], id='mount does not cover the own cgroup'), + ], +) +def test_candidate_dirs( + tmp_path: Path, + mount_root: str, + cgroup_path: str, + expected: list[str], +) -> None: + """Walks from the cgroup of this process up to the top of the mount.""" + hierarchy = cgroup._Hierarchy(point=tmp_path, root=mount_root, own_path=cgroup_path) + + dirs = cgroup._candidate_dirs(hierarchy) + + assert list(dirs) == [tmp_path / relative if relative else tmp_path for relative in expected] + + +@pytest.mark.parametrize( + ('mountinfo', 'self_cgroup', 'files', 'expected'), + [ + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + {'memory.max': '536870912\n', 'memory.current': '1000\n', 'memory.stat': 'inactive_file 400\n'}, + 536870912, + id='v2 limit', + ), + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + {'memory.max': 'max\n', 'memory.current': '1000\n', 'memory.stat': 'inactive_file 400\n'}, + None, + id='v2 unlimited', + ), + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/pod/container'), + { + 'pod/container/memory.max': 'max\n', + 'pod/container/memory.current': '1000\n', + 'pod/container/memory.stat': 'inactive_file 400\n', + 'pod/memory.max': '268435456\n', + 'pod/memory.current': '9000\n', + 'pod/memory.stat': 'inactive_file 1000\n', + }, + 268435456, + id='limit inherited from an ancestor', + ), + pytest.param( + V1_MOUNTINFO, + V1_SELF_CGROUP.format(path='/'), + { + 'memory/memory.limit_in_bytes': '536870912\n', + 'memory/memory.usage_in_bytes': '1000\n', + 'memory/memory.stat': 'total_inactive_file 400\n', + }, + 536870912, + id='v1 limit', + ), + pytest.param( + V1_MOUNTINFO, + V1_SELF_CGROUP.format(path='/'), + # Only a caller that knows the memory of the host can tell the sentinel apart from a real limit. + { + 'memory/memory.limit_in_bytes': '9223372036854771712\n', + 'memory/memory.usage_in_bytes': '1000\n', + 'memory/memory.stat': 'total_inactive_file 400\n', + }, + 9223372036854771712, + id='v1 unlimited sentinel', + ), + ], +) +def test_get_memory_limit( + fake_cgroup: Callable[..., Path], + mountinfo: str, + self_cgroup: str, + files: dict[str, str], + expected: int | None, +) -> None: + """Reads the memory limit under both cgroup interfaces.""" + fake_cgroup(mountinfo=mountinfo, self_cgroup=self_cgroup, files=files) + + memory_limit = cgroup.get_memory_limit() + + assert (memory_limit.limit if memory_limit is not None else None) == expected + + +def test_get_memory_limit_tightest_level(fake_cgroup: Callable[..., Path]) -> None: + """Reports the tightest limit of the chain, so a budget cannot exceed what the kernel enforces.""" + fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/kubepods/pod/container'), + files={ + # The node-level cgroup is generous and busy, because every pod on the node is charged against it. + 'kubepods/memory.max': '8000\n', + 'kubepods/memory.current': '6000\n', + 'kubepods/memory.stat': 'inactive_file 0\n', + # The container this process runs in is limited far more tightly, and barely uses its share. + 'kubepods/pod/container/memory.max': '1000\n', + 'kubepods/pod/container/memory.current': '100\n', + 'kubepods/pod/container/memory.stat': 'inactive_file 0\n', + }, + ) + + memory_limit = cgroup.get_memory_limit() + + assert memory_limit is not None + assert memory_limit.limit == 1000 + assert memory_limit.working_set == 750 + + +def test_get_memory_limit_worst_level(fake_cgroup: Callable[..., Path]) -> None: + """Reports the utilization of the level closest to running out, which is not always the tightest one.""" + fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/qos/pod'), + files={ + # The pod is nearly full of its own generous limit. + 'qos/pod/memory.max': '1000\n', + 'qos/pod/memory.current': '960\n', + 'qos/pod/memory.stat': 'inactive_file 10\n', + # The QoS class above is tighter, but half of what it holds belongs to the sibling pods. + 'qos/memory.max': '500\n', + 'qos/memory.current': '260\n', + 'qos/memory.stat': 'inactive_file 10\n', + }, + ) + + memory_limit = cgroup.get_memory_limit() + + assert memory_limit is not None + assert memory_limit.limit == 500 + assert memory_limit.working_set == 475 + + +def test_get_memory_limit_partial_level(fake_cgroup: Callable[..., Path]) -> None: + """Counts a level towards the ceiling even when the usage next to it cannot be read.""" + fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/kubepods/pod/container'), + files={ + 'kubepods/memory.max': '8000\n', + 'kubepods/memory.current': '6000\n', + 'kubepods/memory.stat': 'inactive_file 0\n', + # The tightest limit, with no page cache metric to derive a working set from. + 'kubepods/pod/container/memory.max': '1000\n', + 'kubepods/pod/container/memory.current': '100\n', + 'kubepods/pod/container/memory.stat': 'anon 100\n', + }, + ) + + memory_limit = cgroup.get_memory_limit() + + assert memory_limit is not None + assert memory_limit.limit == 1000 + assert memory_limit.working_set == 750 + + +@pytest.mark.parametrize( + ('mountinfo', 'self_cgroup', 'files'), + [ + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + { + 'memory.max': '536870912\n', + 'memory.current': '1000\n', + 'memory.stat': 'anon 600\ninactive_file 400\nfile 400\n', + }, + id='v2', + ), + pytest.param( + V1_MOUNTINFO, + V1_SELF_CGROUP.format(path='/'), + { + 'memory/memory.limit_in_bytes': '536870912\n', + 'memory/memory.usage_in_bytes': '1000\n', + 'memory/memory.stat': 'rss 600\ntotal_inactive_file 400\n', + }, + id='v1', + ), + ], +) +def test_get_memory_limit_working_set( + fake_cgroup: Callable[..., Path], + mountinfo: str, + self_cgroup: str, + files: dict[str, str], +) -> None: + """Subtracts the page cache the kernel drops on demand, which would not predict a kill.""" + fake_cgroup(mountinfo=mountinfo, self_cgroup=self_cgroup, files=files) + + memory_limit = cgroup.get_memory_limit() + + assert memory_limit is not None + assert memory_limit.working_set == 600 + + +def test_get_memory_limit_no_working_set( + fake_cgroup: Callable[..., Path], + caplog: pytest.LogCaptureFixture, +) -> None: + """Drops a limit that no usage can be paired with, and warns instead of pairing it with the raw usage.""" + fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/'), + files={'memory.max': '536870912\n', 'memory.current': '1000\n', 'memory.stat': 'anon 600\n'}, + ) + + with caplog.at_level(logging.WARNING, logger=cgroup.logger.name): + assert cgroup.get_memory_limit() is None + + assert [record for record in caplog.records if 'cgroup memory limit' in record.getMessage()] + + +def test_get_memory_limit_missing_files(fake_cgroup: Callable[..., Path]) -> None: + """Reads the closest level that carries the controller, which the own cgroup need not.""" + root = fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/pod/container'), + files={ + 'pod/memory.max': '268435456\n', + 'pod/memory.current': '1000\n', + 'pod/memory.stat': 'inactive_file 400\n', + }, + ) + # The cgroup of the process exists, it just carries no memory files of its own. + (root / 'pod' / 'container').mkdir() + + memory_limit = cgroup.get_memory_limit() + + assert memory_limit is not None + assert memory_limit.limit == 268435456 + assert memory_limit.working_set == 600 + + +@pytest.mark.parametrize( + ('mountinfo', 'self_cgroup', 'files', 'expected'), + [ + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + {'cpu.max': '200000 100000\n', 'cpu.stat': 'usage_usec 0\n'}, + 2.0, + id='v2 quota', + ), + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + {'cpu.max': '50000 100000\n', 'cpu.stat': 'usage_usec 0\n'}, + 0.5, + id='v2 fractional quota', + ), + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + {'cpu.max': 'max 100000\n', 'cpu.stat': 'usage_usec 0\n'}, + None, + id='v2 unlimited', + ), + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/pod/container'), + { + 'pod/container/cpu.max': 'max 100000\n', + 'pod/container/cpu.stat': 'usage_usec 0\n', + 'pod/cpu.max': '150000 100000\n', + }, + 1.5, + id='v2 quota inherited from an ancestor', + ), + pytest.param( + V1_MOUNTINFO, + V1_SELF_CGROUP.format(path='/'), + { + 'cpu,cpuacct/cpu.cfs_quota_us': '150000\n', + 'cpu,cpuacct/cpu.cfs_period_us': '100000\n', + 'cpu,cpuacct/cpuacct.usage': '0\n', + }, + 1.5, + id='v1 quota', + ), + pytest.param( + V1_MOUNTINFO, + V1_SELF_CGROUP.format(path='/'), + { + 'cpu,cpuacct/cpu.cfs_quota_us': '-1\n', + 'cpu,cpuacct/cpu.cfs_period_us': '100000\n', + 'cpu,cpuacct/cpuacct.usage': '0\n', + }, + None, + id='v1 unlimited', + ), + ], +) +def test_get_cpu_quota( + fake_cgroup: Callable[..., Path], + mountinfo: str, + self_cgroup: str, + files: dict[str, str], + expected: float | None, +) -> None: + """Reads the CPU bandwidth quota under both cgroup interfaces.""" + fake_cgroup(mountinfo=mountinfo, self_cgroup=self_cgroup, files=files) + + assert cgroup.get_cpu_quota() == expected + + +@pytest.mark.parametrize( + ('mountinfo', 'self_cgroup', 'files', 'expected'), + [ + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + {'cpuset.cpus.effective': '0-1\n'}, + 2, + id='v2 range', + ), + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + {'cpuset.cpus.effective': '0-1,4,6-7\n'}, + 5, + id='v2 ranges mixed with single cores', + ), + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + {'cpuset.cpus.effective': '\n'}, + None, + id='inherited from the parent', + ), + pytest.param( + V1_MOUNTINFO, + V1_SELF_CGROUP.format(path='/'), + {'cpuset/cpuset.cpus': '0-3\n'}, + 4, + id='v1', + ), + ], +) +def test_get_cpu_set_size( + fake_cgroup: Callable[..., Path], + mountinfo: str, + self_cgroup: str, + files: dict[str, str], + expected: int | None, +) -> None: + """Counts the cores of a CPU set spelled as a mix of ranges and single numbers.""" + fake_cgroup(mountinfo=mountinfo, self_cgroup=self_cgroup, files=files) + + assert cgroup.get_cpu_set_size() == expected + + +@pytest.mark.parametrize( + ('mountinfo', 'self_cgroup', 'files'), + [ + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + # cgroup v2 reports the consumed CPU time in microseconds, among other keys. + {'cpu.stat': 'usage_usec 2500000\nuser_usec 2000000\n'}, + id='v2', + ), + pytest.param( + V1_MOUNTINFO, + V1_SELF_CGROUP.format(path='/'), + # cgroup v1 reports it in nanoseconds, in a file of its own. + {'cpu,cpuacct/cpuacct.usage': '2500000000\n'}, + id='v1', + ), + ], +) +def test_get_cpu_usage( + fake_cgroup: Callable[..., Path], + mountinfo: str, + self_cgroup: str, + files: dict[str, str], +) -> None: + """Reports the consumed CPU time in seconds under both cgroup interfaces.""" + fake_cgroup(mountinfo=mountinfo, self_cgroup=self_cgroup, files=files) + + assert cgroup.get_cpu_usage() == 2.5 + + +def test_get_controllers_hybrid(fake_cgroup: Callable[..., Path]) -> None: + """Falls back to cgroup v1 for a controller the unified hierarchy does not carry.""" + fake_cgroup( + mountinfo=f'{V2_MOUNTINFO}\n{V1_MOUNTINFO}', + self_cgroup=f'{V2_SELF_CGROUP.format(path="/")}2:memory:/\n', + files={ + 'memory/memory.limit_in_bytes': '536870912\n', + 'memory/memory.usage_in_bytes': '1000\n', + 'memory/memory.stat': 'total_inactive_file 400\n', + }, + ) + + memory = cgroup._get_controllers().memory + memory_limit = cgroup.get_memory_limit() + + assert memory is not None + assert memory.is_v2 is False + assert memory_limit is not None + assert memory_limit.limit == 536870912 + assert memory_limit.working_set == 600 + + +@pytest.mark.usefixtures('_no_cgroup') +def test_no_cgroups() -> None: + """Reports nothing on a system that has no cgroups.""" + assert cgroup.get_memory_limit() is None + assert cgroup.get_cpu_quota() is None + assert cgroup.get_cpu_set_size() is None + assert cgroup.get_cpu_usage() is None + + +# The tests below cover how `system.py` reports the metrics read above, which is the only place they are consumed. + + +@pytest.mark.usefixtures('_fixed_host_memory') +def test_get_memory_info_limited(fake_cgroup: Callable[..., Path]) -> None: + """Reports the limit of the container, not the memory of the host the autoscaler would size its budget from.""" + fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/'), + files={ + 'memory.max': '536870912\n', + 'memory.current': '150000000\n', + 'memory.stat': 'inactive_file 50000000\n', + }, + ) + + memory_info = get_memory_info() + + assert memory_info.total_size == ByteSize(536870912) + assert memory_info.system_wide_used_size == ByteSize(100000000) + + +@pytest.mark.usefixtures('_fixed_host_memory') +def test_get_memory_info_unlimited(fake_cgroup: Callable[..., Path]) -> None: + """Reports the host when the limit is at or above its memory, which is how an unconstrained cgroup spells it.""" + fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/'), + files={ + 'memory.max': f'{HOST_TOTAL_BYTES * 2}\n', + 'memory.current': '150000000\n', + 'memory.stat': 'inactive_file 50000000\n', + }, + ) + + memory_info = get_memory_info() + + assert memory_info.total_size == ByteSize(HOST_TOTAL_BYTES) + assert memory_info.system_wide_used_size == ByteSize(HOST_TOTAL_BYTES - HOST_AVAILABLE_BYTES) + + +@pytest.mark.usefixtures('_fixed_host_memory') +def test_get_memory_info_no_working_set(fake_cgroup: Callable[..., Path]) -> None: + """Falls back to the host as a whole, because a cgroup limit next to host-wide usage compares two scopes.""" + fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/'), + files={'memory.max': '536870912\n', 'memory.current': '150000000\n', 'memory.stat': 'anon 1\n'}, + ) + + memory_info = get_memory_info() + + assert memory_info.total_size == ByteSize(HOST_TOTAL_BYTES) + assert memory_info.system_wide_used_size == ByteSize(HOST_TOTAL_BYTES - HOST_AVAILABLE_BYTES) + + +@pytest.mark.usefixtures('_one_second_per_sample') +def test_get_cpu_info_quota(fake_cgroup: Callable[..., Path], monkeypatch: pytest.MonkeyPatch) -> None: + """Measures the CPU against the bandwidth quota.""" + root = fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/'), + files={'cpu.max': '200000 100000\n', 'cpu.stat': 'usage_usec 0\n'}, + ) + # A host-wide reading would show a fully loaded machine, so it is visible if the fallback is taken by mistake. + monkeypatch.setattr(psutil, 'cpu_percent', lambda **_: 100.0) + + # A counter that only grows needs two readings, which the first sample waits for instead of falling back. + assert get_cpu_info().used_ratio == 0.0 + + (root / 'cpu.stat').write_text('usage_usec 1000000\n') + + # One core-second over one second of wall time, out of the two cores the quota allows. + assert get_cpu_info().used_ratio == pytest.approx(0.5) + + +@pytest.mark.usefixtures('_one_second_per_sample') +def test_get_cpu_info_cpu_set(fake_cgroup: Callable[..., Path], monkeypatch: pytest.MonkeyPatch) -> None: + """Measures the CPU against a set that restricts the cores without setting any quota.""" + root = fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/'), + files={'cpuset.cpus.effective': '0-1\n', 'cpu.stat': 'usage_usec 0\n'}, + ) + monkeypatch.setattr(psutil, 'cpu_count', lambda: 8) + + assert get_cpu_info().used_ratio == 0.0 + + (root / 'cpu.stat').write_text('usage_usec 2000000\n') + + # Two core-seconds over one second of wall time saturate the two cores of the set. + assert get_cpu_info().used_ratio == pytest.approx(1.0) + + +@pytest.mark.usefixtures('_one_second_per_sample') +def test_get_cpu_info_counter_restart(fake_cgroup: Callable[..., Path], monkeypatch: pytest.MonkeyPatch) -> None: + """Clamps the counter that restarts when the process is moved to another cgroup.""" + root = fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/'), + files={'cpu.max': '200000 100000\n', 'cpu.stat': 'usage_usec 5000000\n'}, + ) + monkeypatch.setattr(psutil, 'cpu_percent', lambda **_: 100.0) + + get_cpu_info() + (root / 'cpu.stat').write_text('usage_usec 0\n') + + assert get_cpu_info().used_ratio == 0.0 + + +@pytest.mark.usefixtures('_one_second_per_sample') +def test_get_cpu_info_failed_reading(fake_cgroup: Callable[..., Path], monkeypatch: pytest.MonkeyPatch) -> None: + """Keeps the earlier reading when the next one fails, because a counter that only grows stays comparable.""" + root = fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/'), + files={'cpu.max': '200000 100000\n', 'cpu.stat': 'usage_usec 0\n'}, + ) + monkeypatch.setattr(psutil, 'cpu_percent', lambda **_: 100.0) + + assert get_cpu_info().used_ratio == 0.0 + + (root / 'cpu.stat').unlink() + assert get_cpu_info().used_ratio == 1.0 + + (root / 'cpu.stat').write_text('usage_usec 1000000\n') + + # Starting over instead would compare the reading against itself and report nothing consumed. + assert get_cpu_info().used_ratio == pytest.approx(0.5) + + +def test_get_cpu_info_no_usage( + fake_cgroup: Callable[..., Path], + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """Warns when a CPU limit is found but nothing can be measured against it.""" + fake_cgroup( + mountinfo=V2_MOUNTINFO, + self_cgroup=V2_SELF_CGROUP.format(path='/'), + files={'cpu.max': '200000 100000\n'}, + ) + monkeypatch.setattr(psutil, 'cpu_percent', lambda **_: 42.0) + + with caplog.at_level(logging.WARNING, logger=system.logger.name): + assert get_cpu_info().used_ratio == pytest.approx(0.42) + + assert [record for record in caplog.records if 'cgroup CPU limit' in record.getMessage()] + + +@pytest.mark.usefixtures('_no_cgroup') +def test_get_cpu_info_no_limit(monkeypatch: pytest.MonkeyPatch) -> None: + """Falls back to the load of the whole machine, which is what matters when nothing restricts the CPU.""" + monkeypatch.setattr(psutil, 'cpu_percent', lambda **_: 42.0) + + assert get_cpu_info().used_ratio == pytest.approx(0.42) + + +@pytest.mark.parametrize( + ('quota', 'cpu_set_cores', 'host_cores', 'expected'), + [ + pytest.param(None, None, 8, None, id='nothing restricts the cpu'), + pytest.param(2.0, None, 8, 2.0, id='bandwidth quota only'), + pytest.param(None, 2, 8, 2.0, id='cpu set only'), + pytest.param(None, 8, 8, None, id='a cpu set covering every core is not a restriction'), + pytest.param(4.0, 2, 8, 2.0, id='cpu set is tighter than the quota'), + pytest.param(1.0, 2, 8, 1.0, id='quota is tighter than the cpu set'), + ], +) +def test_get_allowed_cpu_cores( + monkeypatch: pytest.MonkeyPatch, + quota: float | None, + cpu_set_cores: int | None, + host_cores: int, + expected: float | None, +) -> None: + """Takes the tighter of the bandwidth quota and the CPU set, which restrict the CPU independently.""" + monkeypatch.setattr(cgroup, 'get_cpu_quota', lambda: quota) + monkeypatch.setattr(cgroup, 'get_cpu_set_size', lambda: cpu_set_cores) + monkeypatch.setattr(psutil, 'cpu_count', lambda: host_cores) + + assert system._get_allowed_cpu_cores() == expected From cebb3fdd95ba665a11f55fc8e60ac30b2ca91adc Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Thu, 6 Aug 2026 23:12:33 +0000 Subject: [PATCH 2/2] fix --- src/crawlee/_utils/cgroup.py | 17 ++++++++++++++-- tests/unit/_utils/test_cgroup.py | 34 ++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/src/crawlee/_utils/cgroup.py b/src/crawlee/_utils/cgroup.py index f094b1bbfc..2fcf90c3db 100644 --- a/src/crawlee/_utils/cgroup.py +++ b/src/crawlee/_utils/cgroup.py @@ -333,6 +333,11 @@ def _read_hierarchies() -> tuple[_Hierarchy | None, dict[str, _Hierarchy]]: except ValueError: continue + # `/proc/self/cgroup` spells the same paths unescaped, so without this the two cannot be compared, and the + # mount point cannot be opened either. + mount_root = _unescape(mount_root) + mount_point = _unescape(mount_point) + if filesystem == 'cgroup2' and unified_path is not None: # The same hierarchy can be bind-mounted a second time, for instance by an agent that watches the host # from inside a container. Mounts are listed in the order they were made, so the first one is ours. @@ -375,14 +380,22 @@ def _read_own_paths() -> tuple[str | None, dict[str, str]]: return unified, controllers +def _unescape(field: str) -> str: + """Decode the octal sequences a path field of `/proc/self/mountinfo` escapes special characters as.""" + # The backslash goes last. Undoing it first would decode `\134040`, an escaped backslash followed by `040`, + # into a space. + return field.replace('\\040', ' ').replace('\\011', '\t').replace('\\012', '\n').replace('\\134', '\\') + + def _count_cpu_list(cpu_list: str) -> int | None: """Count the CPUs a control file lists as a mix of ranges and single numbers, e.g. `0-3,8`.""" count = 0 try: for part in cpu_list.split(','): - first, separator, last = part.partition('-') - count += int(last) - int(first) + 1 if separator else 1 + start, _separator, end = part.partition('-') + # A single core carries no end, so it counts as a range of one. + count += int(end or start) - int(start) + 1 except ValueError: return None diff --git a/tests/unit/_utils/test_cgroup.py b/tests/unit/_utils/test_cgroup.py index 8d0dc9a6c8..b2531670bf 100644 --- a/tests/unit/_utils/test_cgroup.py +++ b/tests/unit/_utils/test_cgroup.py @@ -138,6 +138,33 @@ def test_read_hierarchies_bad_lines(fake_cgroup: Callable[..., Path]) -> None: assert unified.point == root +@pytest.mark.parametrize( + ('escaped', 'expected'), + [ + pytest.param('/plain/path', '/plain/path', id='nothing to decode'), + pytest.param('/mnt\\040point', '/mnt point', id='space'), + pytest.param('/tab\\011here', '/tab\there', id='tab'), + pytest.param('/back\\134slash', '/back\\slash', id='backslash'), + pytest.param('/literal\\134040', '/literal\\040', id='escaped backslash in front of an octal sequence'), + ], +) +def test_unescape(escaped: str, expected: str) -> None: + """Decodes the octal sequences a path field of the mount table escapes special characters as.""" + assert cgroup._unescape(escaped) == expected + + +def test_read_hierarchies_escaped_paths(fake_cgroup: Callable[..., Path]) -> None: + """Decodes both path fields, which `/proc/self/cgroup` spells unescaped and so cannot be compared against.""" + mountinfo = '25 30 0:22 /docker\\040abc {root}/mnt\\040point rw shared:4 - cgroup2 cgroup2 rw' + root = fake_cgroup(mountinfo=mountinfo, self_cgroup=V2_SELF_CGROUP.format(path='/docker abc'), files={}) + + unified, _controllers = cgroup._read_hierarchies() + + assert unified is not None + assert unified.point == root / 'mnt point' + assert unified.root == '/docker abc' + + @pytest.mark.parametrize( ('self_cgroup', 'expected_unified', 'expected_controllers'), [ @@ -523,6 +550,13 @@ def test_get_cpu_quota( 4, id='v1', ), + pytest.param( + V2_MOUNTINFO, + V2_SELF_CGROUP.format(path='/'), + {'cpuset.cpus.effective': '0-1,nonsense\n'}, + None, + id='every entry has to parse, not just the ranges', + ), ], ) def test_get_cpu_set_size(