From 165ea86bf88ca6967055469c2a97513e74312d86 Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Tue, 8 Sep 2026 21:26:04 +0000 Subject: [PATCH] respect container memory and CPU limits platform --- docs/guides/scaling_crawlers.mdx | 6 + pyproject.toml | 5 + src/crawlee/_utils/system.py | 90 +++++++++++-- tests/unit/_utils/test_system.py | 211 ++++++++++++++++++++++++++++++- uv.lock | 12 +- 5 files changed, 312 insertions(+), 12 deletions(-) diff --git a/docs/guides/scaling_crawlers.mdx b/docs/guides/scaling_crawlers.mdx index 152d852e60..e37ea8b7d0 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, a systemd slice and a Windows job object each carry a limit of their own. Crawlee reads the limit that applies to the process 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. 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/pyproject.toml b/pyproject.toml index da4ecd05bc..e7bca5ad07 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,7 @@ keywords = [ dependencies = [ "async-timeout>=5.0.1", "cachetools>=5.5.0", + "cgroups-sensor>=0.1.0,<1.0.0", "colorama>=0.4.0", "impit>=0.13.2", "more-itertools>=10.2.0", @@ -316,8 +317,12 @@ exclude-newer = "24 hours" apify-client = false apify-shared = false apify_fingerprint_datapoints = false +cgroups-sensor = false crawlee = false +[tool.uv.sources] +cgroups-sensor = { git = "https://github.com/apify/cgroups-sensor.git"} + # Run tasks with: uv run poe [tool.poe.tasks] clean = "rm -rf .coverage .pytest_cache .ruff_cache .ty_cache .uv-cache build coverage-unit.xml dist htmlcov website/.docusaurus website/module_shortcuts.json website/node_modules " diff --git a/src/crawlee/_utils/system.py b/src/crawlee/_utils/system.py index 45d0483679..3dc4ea324a 100644 --- a/src/crawlee/_utils/system.py +++ b/src/crawlee/_utils/system.py @@ -2,10 +2,12 @@ import os import sys +import threading from datetime import datetime, timezone from logging import WARNING, getLogger from typing import TYPE_CHECKING, Annotated +import cgroups_sensor import psutil from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, PlainValidator @@ -19,6 +21,12 @@ # 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 fallback measures for. A window shorter than 0.01 seconds is refused by the sensor.""" + +_cpu_load = cgroups_sensor.CpuLoad() +"""Process-wide CPU sampler, measuring across the gap between calls.""" + class _PssAvailability: """Process-wide latch for whether the PSS memory metric exists on this system at all. @@ -185,7 +193,10 @@ 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. + + Under a container limit this is the limit rather than the memory of the host machine. + """ system_wide_used_size: Annotated[ ByteSize, @@ -193,27 +204,72 @@ 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. + + Under a container limit this is the memory charged against that limit, as `docker stats` reports it. + """ + + +class _ResourceLimits: + """Process-wide latch keeping the limits report to one line per process, rather than one per sample.""" + + is_pending = True + lock = threading.Lock() + + +def _log_resource_limits() -> None: + """Report the limits applying to this process, at most once per process and only where any apply.""" + # The latch is consumed before the reading, so a sensor that raises costs one snapshot rather than every one. + with _ResourceLimits.lock: + if not _ResourceLimits.is_pending: + return + _ResourceLimits.is_pending = False + + limits = cgroups_sensor.snapshot() + cores = limits.cpu_limit + + # An unrestricted process is the ordinary case, and a line saying so explains nothing. + if limits.memory_budget is None and cores is None: + return + + memory = str(ByteSize(limits.memory_budget.limit)) if limits.memory_budget else 'unrestricted' + cpu = f'{cores:g} core{"" if cores == 1 else "s"}' if cores is not None else 'unrestricted' + logger.info(f'Resource limits applying to this process: memory {memory}, CPU {cpu}.') 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. + Under a container limit the load is measured against the cores this process may use. The sampler measures across + the gap between calls, so the first sample of the process falls back to a short measurement of its own. Without a + limit the process competes for the whole machine, and `psutil.cpu_percent()` answers instead. """ logger.debug('Calling get_cpu_info()...') - cpu_percent = psutil.cpu_percent(interval=0.1) - return CpuInfo(used_ratio=cpu_percent / 100) + + # Read on every sample rather than latched, because a limit can be resized while the process runs. + if cgroups_sensor.get_cpu_limit() is None: + return CpuInfo(used_ratio=psutil.cpu_percent(interval=_CPU_SAMPLE_INTERVAL_SECS) / 100) + + used_ratio = _cpu_load.sample() + + if used_ratio is None: + used_ratio = cgroups_sensor.get_cpu_used_ratio(_CPU_SAMPLE_INTERVAL_SECS) + + 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 limit applying to this process whenever one restricts how much memory it may use. """ logger.debug('Calling get_memory_info()...') + _log_resource_limits() current_process = psutil.Process(os.getpid()) # Retrieve estimated memory usage of the current process. Deliberately not guarded - a process can always read @@ -237,9 +293,25 @@ 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]: + """Get the total and the used memory to report, narrowed to the limit applying to this process.""" + budget = cgroups_sensor.get_memory_budget() + + if budget is None: + return host_total_bytes, host_used_bytes + + # Not clamped to the memory of the machine: a Windows job limits commit, so that would pair a commit charge with + # a physical ceiling. + return budget.limit, budget.used diff --git a/tests/unit/_utils/test_system.py b/tests/unit/_utils/test_system.py index 287de5000c..bb4cf9ebec 100644 --- a/tests/unit/_utils/test_system.py +++ b/tests/unit/_utils/test_system.py @@ -2,12 +2,14 @@ import logging import sys +import threading from multiprocessing import get_context, synchronize from multiprocessing.shared_memory import SharedMemory from types import SimpleNamespace from typing import TYPE_CHECKING from unittest.mock import Mock +import cgroups_sensor import psutil import pytest @@ -19,6 +21,9 @@ if TYPE_CHECKING: from collections.abc import Callable +HOST_TOTAL_BYTES = 8 * 1024**3 +HOST_AVAILABLE_BYTES = 3 * 1024**3 + class FakeProcess: """Stand-in for `psutil.Process` that lets a test decide what a child process reports as its memory usage.""" @@ -65,9 +70,28 @@ def fill_buffer(buffer: memoryview, size: int) -> None: @pytest.fixture(autouse=True) def _isolated_module_state(monkeypatch: pytest.MonkeyPatch) -> None: - """Reset the process-wide state of the module, so that dedup keys and the PSS latch do not leak between tests.""" + """Reset the process-wide state of the module, so that dedup keys and the latches do not leak between tests.""" monkeypatch.setattr(system, 'logger_once', LoggerOnce(system.logger)) monkeypatch.setattr(system._PssAvailability, 'is_available', True) + monkeypatch.setattr(system._ResourceLimits, 'is_pending', True) + + +@pytest.fixture(autouse=True) +def cpu_load(monkeypatch: pytest.MonkeyPatch) -> Mock: + """Replace the CPU readings taken against a limit, so that neither a leaked reading nor a real limit is measured.""" + sampler = Mock(spec=cgroups_sensor.CpuLoad) + # What all three report where nothing restricts the CPU, which sends `get_cpu_info` to the psutil fallback. + sampler.sample.return_value = None + monkeypatch.setattr(system, '_cpu_load', sampler) + monkeypatch.setattr(cgroups_sensor, 'get_cpu_used_ratio', Mock(return_value=None)) + monkeypatch.setattr(cgroups_sensor, 'get_cpu_limit', Mock(return_value=None)) + return sampler + + +@pytest.fixture +def _cpu_limited(monkeypatch: pytest.MonkeyPatch) -> None: + """Report a CPU limit, so that the load is measured against it rather than against the host machine.""" + monkeypatch.setattr(cgroups_sensor, 'get_cpu_limit', Mock(return_value=1.0)) @pytest.fixture @@ -77,6 +101,23 @@ def measured_current_process(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(psutil.Process, 'memory_info', lambda _process: SimpleNamespace(rss=100)) +@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', + Mock(return_value=SimpleNamespace(total=HOST_TOTAL_BYTES, available=HOST_AVAILABLE_BYTES)), + ) + + +def fake_snapshot( + *, memory_budget: cgroups_sensor.MemoryBudget | None = None, cpu_limit: float | None = None +) -> cgroups_sensor.Snapshot: + """Stand in for `cgroups_sensor.snapshot()`, describing an unrestricted process unless told otherwise.""" + return cgroups_sensor.Snapshot(memory_budget=memory_budget, cpu_limit=cpu_limit, cpu_usage=None) + + def test_get_memory_info_returns_valid_values() -> None: memory_info = get_memory_info() @@ -211,6 +252,174 @@ def test_get_cpu_info_returns_valid_values() -> None: assert 0 <= cpu_info.used_ratio <= 1 +@pytest.mark.usefixtures('_fixed_host_memory', 'measured_current_process') +def test_get_memory_info_reports_the_limit(monkeypatch: pytest.MonkeyPatch) -> None: + """A limit applying to the process replaces the memory of the host machine.""" + budget = cgroups_sensor.MemoryBudget(limit=512 * 1024**2, used=100 * 1024**2, available=412 * 1024**2) + monkeypatch.setattr(cgroups_sensor, 'get_memory_budget', Mock(return_value=budget)) + + memory_info = get_memory_info() + + assert memory_info.total_size == ByteSize(budget.limit) + assert memory_info.system_wide_used_size == ByteSize(budget.used) + + +@pytest.mark.usefixtures('_fixed_host_memory', 'measured_current_process') +def test_get_memory_info_falls_back_to_the_host(monkeypatch: pytest.MonkeyPatch) -> None: + """An unrestricted process is measured against the memory of the host machine.""" + monkeypatch.setattr(cgroups_sensor, 'get_memory_budget', Mock(return_value=None)) + + 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('_cpu_limited') +def test_get_cpu_info_measures_against_the_limit(monkeypatch: pytest.MonkeyPatch, cpu_load: Mock) -> None: + """A sampled load is reported as it is, without measuring the host machine as well.""" + cpu_load.sample.return_value = 0.5 + cpu_percent = Mock(return_value=42.0) + monkeypatch.setattr(psutil, 'cpu_percent', cpu_percent) + + assert get_cpu_info().used_ratio == 0.5 + cpu_percent.assert_not_called() + + +@pytest.mark.usefixtures('_cpu_limited') +def test_get_cpu_info_measures_a_window_when_the_sampler_has_no_reading( + monkeypatch: pytest.MonkeyPatch, cpu_load: Mock +) -> None: + """A sampler with nothing to report yet is covered by a short measurement against the same limit.""" + get_cpu_used_ratio = Mock(return_value=0.25) + monkeypatch.setattr(cgroups_sensor, 'get_cpu_used_ratio', get_cpu_used_ratio) + cpu_percent = Mock(return_value=42.0) + monkeypatch.setattr(psutil, 'cpu_percent', cpu_percent) + + assert get_cpu_info().used_ratio == 0.25 + get_cpu_used_ratio.assert_called_once_with(system._CPU_SAMPLE_INTERVAL_SECS) + # The measurement is refused below 0.01 seconds and nothing on the path catches that, so a window this short + # would raise in every limited container while a mocked measurement stays happy with it. + assert system._CPU_SAMPLE_INTERVAL_SECS >= 0.01 + cpu_percent.assert_not_called() + cpu_load.sample.assert_called_once() + + +@pytest.mark.parametrize( + ('sampled', 'measured'), + [ + pytest.param(0.0, None, id='sampled'), + pytest.param(None, 0.0, id='measured over a window'), + ], +) +@pytest.mark.usefixtures('_cpu_limited') +def test_get_cpu_info_reports_an_idle_limit_as_no_load( + monkeypatch: pytest.MonkeyPatch, cpu_load: Mock, sampled: float | None, measured: float | None +) -> None: + """An idle limited container reports no load, which is a reading of zero rather than a missing one.""" + cpu_load.sample.return_value = sampled + monkeypatch.setattr(cgroups_sensor, 'get_cpu_used_ratio', Mock(return_value=measured)) + cpu_percent = Mock(return_value=42.0) + monkeypatch.setattr(psutil, 'cpu_percent', cpu_percent) + + assert get_cpu_info().used_ratio == 0.0 + cpu_percent.assert_not_called() + + +def test_get_cpu_info_measures_the_host_without_a_limit(monkeypatch: pytest.MonkeyPatch, cpu_load: Mock) -> None: + """Without a limit the process competes for the whole machine, and nothing is measured against a limit.""" + get_cpu_used_ratio = Mock(return_value=0.5) + monkeypatch.setattr(cgroups_sensor, 'get_cpu_used_ratio', get_cpu_used_ratio) + monkeypatch.setattr(psutil, 'cpu_percent', Mock(return_value=42.0)) + + assert get_cpu_info().used_ratio == 0.42 + cpu_load.sample.assert_not_called() + get_cpu_used_ratio.assert_not_called() + + +@pytest.mark.parametrize( + ('memory_budget', 'cpu_limit', 'expected_message'), + [ + pytest.param(None, None, None, id='unrestricted'), + pytest.param( + cgroups_sensor.MemoryBudget(limit=512 * 1024**2, used=100 * 1024**2, available=412 * 1024**2), + 1.0, + 'memory 512.00 MB, CPU 1 core.', + id='single core', + ), + pytest.param(None, 2.5, 'memory unrestricted, CPU 2.5 cores.', id='fractional cores'), + ], +) +@pytest.mark.usefixtures('measured_current_process') +def test_log_resource_limits_reports_what_applies( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + memory_budget: cgroups_sensor.MemoryBudget | None, + cpu_limit: float | None, + expected_message: str | None, +) -> None: + """A limit that applies is reported as one line, and an unrestricted process is not reported at all.""" + snapshot = fake_snapshot(memory_budget=memory_budget, cpu_limit=cpu_limit) + monkeypatch.setattr(cgroups_sensor, 'snapshot', Mock(return_value=snapshot)) + + with caplog.at_level(logging.INFO, logger=system.logger.name): + get_memory_info() + + reported = [record.getMessage() for record in caplog.records if 'Resource limits' in record.getMessage()] + + if expected_message is None: + assert not reported + else: + assert [message for message in reported if expected_message in message] + + +@pytest.mark.usefixtures('measured_current_process') +def test_log_resource_limits_reports_once(monkeypatch: pytest.MonkeyPatch) -> None: + """Reading the limits walks the whole hierarchy, so it is not repeated on every sample.""" + snapshot = Mock(return_value=fake_snapshot()) + monkeypatch.setattr(cgroups_sensor, 'snapshot', snapshot) + + get_memory_info() + get_memory_info() + + snapshot.assert_called_once() + + +@pytest.mark.usefixtures('measured_current_process') +def test_log_resource_limits_lets_a_failing_sensor_surface(monkeypatch: pytest.MonkeyPatch) -> None: + """A sensor that raises is not swallowed, and the latch keeps it to the first sample.""" + snapshot = Mock(side_effect=RuntimeError('Nothing to read here.')) + monkeypatch.setattr(cgroups_sensor, 'snapshot', snapshot) + + with pytest.raises(RuntimeError): + get_memory_info() + + # The latch is consumed first, so the next sample reports as usual rather than raising again. + assert get_memory_info().current_size >= ByteSize(100) + snapshot.assert_called_once() + + +def test_log_resource_limits_reports_once_when_two_threads_race(monkeypatch: pytest.MonkeyPatch) -> None: + """Two event managers sampling in their own threads report the limits once between them, not once each.""" + snapshot = Mock(return_value=fake_snapshot()) + monkeypatch.setattr(cgroups_sensor, 'snapshot', snapshot) + barrier = threading.Barrier(parties=2) + + def report() -> None: + barrier.wait() + system._log_resource_limits() + + threads = [threading.Thread(target=report) for _ in range(2)] + + for thread in threads: + thread.start() + + for thread in threads: + thread.join() + + snapshot.assert_called_once() + + # The estimation is asserted on absolute memory readings, which hold only as long as nothing else on the machine makes # the kernel reclaim the pages allocated below. Running alongside the other test workers is enough to break that. @pytest.mark.run_alone diff --git a/uv.lock b/uv.lock index e6d16b5b61..8668c07247 100644 --- a/uv.lock +++ b/uv.lock @@ -14,10 +14,11 @@ exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for exclude-newer-span = "PT24H" [options.exclude-newer-package] +apify-client = false apify-fingerprint-datapoints = false -crawlee = false apify-shared = false -apify-client = false +cgroups-sensor = false +crawlee = false [[package]] name = "aiomysql" @@ -537,6 +538,11 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/db/3c/33bac158f8ab7f89b2e59426d5fe2e4f63f7ed25df84c036890172b412b5/cfgv-3.5.0-py2.py3-none-any.whl", hash = "sha256:a8dc6b26ad22ff227d2634a65cb388215ce6cc96bbcc5cfde7641ae87e8dacc0", size = 7445, upload-time = "2025-11-19T20:55:50.744Z" }, ] +[[package]] +name = "cgroups-sensor" +version = "0.1.0" +source = { git = "https://github.com/apify/cgroups-sensor.git#c1772412225b47f7f1138a9107b5266881353aaf" } + [[package]] name = "charset-normalizer" version = "3.5.1" @@ -880,6 +886,7 @@ source = { editable = "." } dependencies = [ { name = "async-timeout" }, { name = "cachetools" }, + { name = "cgroups-sensor" }, { name = "colorama" }, { name = "impit" }, { name = "more-itertools" }, @@ -1044,6 +1051,7 @@ requires-dist = [ { name = "browserforge", marker = "extra == 'playwright'", specifier = ">=1.2.3" }, { name = "browserforge", marker = "extra == 'stagehand'", specifier = ">=1.2.3" }, { name = "cachetools", specifier = ">=5.5.0" }, + { name = "cgroups-sensor", git = "https://github.com/apify/cgroups-sensor.git" }, { name = "colorama", specifier = ">=0.4.0" }, { name = "cookiecutter", marker = "extra == 'cli'", specifier = ">=2.6.0" }, { name = "crawlee", extras = ["adaptive-crawler", "pydantic-ai", "beautifulsoup", "cli", "curl-impersonate", "httpx", "parsel", "playwright", "otel", "sql-sqlite", "sql-postgres", "sql-mysql", "stagehand", "redis"], marker = "extra == 'all'" },