diff --git a/docs/guides/scaling_crawlers.mdx b/docs/guides/scaling_crawlers.mdx index 243ff11c08..71dc399107 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 keeps `min_concurrency` tasks running even while the system is overloaded, and starts additional 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 can each carry a limit of their own. Crawlee reads the limit that applies to the process and scales against it, with nothing to configure. 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. While the default ratio applies under a limit, Crawlee logs the resulting budget as a warning. Setting either `available_memory_ratio` or `memory_mbytes` silences the warning. Whatever the budget, the crawler also throttles once the memory charged against the limit goes above 97% of it, which includes memory used by other processes under the same limit. diff --git a/pyproject.toml b/pyproject.toml index 0ac5cf223b..d083e9a259 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,7 @@ dependencies = [ "colorama>=0.4.0", "impit>=0.13.2", "more-itertools>=10.2.0", + "proclimits>=0.2.0", "protego>=0.5.0", "psutil>=6.0.0", "pydantic-settings>=2.12.0", diff --git a/src/crawlee/_autoscaling/_types.py b/src/crawlee/_autoscaling/_types.py index f321214313..7f78b3a90b 100644 --- a/src/crawlee/_autoscaling/_types.py +++ b/src/crawlee/_autoscaling/_types.py @@ -97,13 +97,13 @@ class MemorySnapshot: """Memory usage of the current Python process and its children.""" system_wide_used_size: ByteSize | None - """Memory usage of all processes, system-wide.""" + """Memory usage of all processes, within the scope `system_wide_memory_size` covers.""" max_memory_size: ByteSize """The maximum memory that can be used by `AutoscaledPool`.""" system_wide_memory_size: ByteSize | None - """Total memory available in the whole system.""" + """Total memory available to this process, which is the memory limit where one applies.""" max_used_memory_ratio: float """The maximum acceptable ratio of `current_size` to `max_memory_size`.""" diff --git a/src/crawlee/_autoscaling/snapshotter.py b/src/crawlee/_autoscaling/snapshotter.py index 89581f062c..457eb11500 100644 --- a/src/crawlee/_autoscaling/snapshotter.py +++ b/src/crawlee/_autoscaling/snapshotter.py @@ -7,6 +7,8 @@ from logging import WARNING, getLogger from typing import TYPE_CHECKING, TypeVar, cast +import proclimits + from crawlee import service_locator from crawlee._autoscaling._types import ClientSnapshot, CpuSnapshot, EventLoopSnapshot, MemorySnapshot, Ratio, Snapshot from crawlee._utils.byte_size import ByteSize @@ -128,6 +130,19 @@ def from_config(cls, config: Configuration | None = None) -> Snapshotter: else Ratio(value=config.available_memory_ratio) ) + # The default ratio protects the machine from the crawler. Under a limit set outside of Crawlee it stacks on + # top of that limit, which is rarely what the user meant. + if not config.memory_mbytes and 'available_memory_ratio' not in config.model_fields_set: + budget = proclimits.get_memory_budget() + if budget is not None: + logger_once.log( + f'Setting max memory of this run to {config.available_memory_ratio:.0%} of the ' + f'{ByteSize(budget.limit)} memory limit applying to this process. Use the CRAWLEE_MEMORY_MBYTES ' + 'or CRAWLEE_AVAILABLE_MEMORY_RATIO environment variable to override it.', + key='default_memory_ratio_under_limit', + level=WARNING, + ) + return cls( max_used_cpu_ratio=config.max_used_cpu_ratio, max_used_memory_ratio=config.max_used_memory_ratio, diff --git a/src/crawlee/_utils/system.py b/src/crawlee/_utils/system.py index 45d0483679..794bc77534 100644 --- a/src/crawlee/_utils/system.py +++ b/src/crawlee/_utils/system.py @@ -6,6 +6,7 @@ from logging import WARNING, getLogger from typing import TYPE_CHECKING, Annotated +import proclimits import psutil from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, PlainValidator @@ -19,6 +20,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 blocking CPU measurement lasts. A window shorter than 0.01 seconds is refused by the sensor.""" + class _PssAvailability: """Process-wide latch for whether the PSS memory metric exists on this system at all. @@ -185,7 +189,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 +200,74 @@ 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. + """ + + +class _ResourceLimits: + """Process-wide latch keeping the limits report to one line per process, rather than one per sample.""" + + is_pending = True + + +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. + if not _ResourceLimits.is_pending: + return + + _ResourceLimits.is_pending = False + limits = proclimits.snapshot() + cores = limits.cpu_limit -def get_cpu_info() -> CpuInfo: + 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(cpu_load: proclimits.CpuLoad | None = None) -> 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, and a call it has no reading for, such as the first, 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. + + Args: + cpu_load: The sampler owned by the caller. Two callers sharing one would measure each other's windows. + Without one, every call under a limit takes a short measurement of its own. """ 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 proclimits.get_cpu_limit() is None: + return CpuInfo(used_ratio=psutil.cpu_percent(interval=_CPU_SAMPLE_INTERVAL_SECS) / 100) + + used_ratio = cpu_load.sample() if cpu_load is not None else None + + if used_ratio is None: + used_ratio = proclimits.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 @@ -236,10 +290,18 @@ def get_memory_info() -> MemoryInfo: for child in children: current_size_bytes += _get_child_used_memory(child) - vm = psutil.virtual_memory() + budget = proclimits.get_memory_budget() + + if budget is None: + vm = psutil.virtual_memory() + total_size_bytes, system_wide_used_size_bytes = vm.total, vm.total - vm.available + else: + # Not clamped to the memory of the machine: a Windows job limits commit, so that would pair a commit charge + # with a physical ceiling. + total_size_bytes, system_wide_used_size_bytes = budget.limit, budget.used 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), ) diff --git a/src/crawlee/configuration.py b/src/crawlee/configuration.py index e5f45710d1..0da2373c05 100644 --- a/src/crawlee/configuration.py +++ b/src/crawlee/configuration.py @@ -186,9 +186,9 @@ class Configuration(BaseSettings): le=1.0, ), ] = 0.25 - """The maximum proportion of system memory to use. If `memory_mbytes` is not provided, this ratio is used to - calculate the maximum memory. This option is utilized by the `Snapshotter` and supports the dynamic system memory - scaling.""" + """The maximum proportion of the memory available to this process to use, which is the memory limit where one + applies. If `memory_mbytes` is not provided, this ratio is used to calculate the maximum memory. This option is + utilized by the `Snapshotter` and supports the dynamic system memory scaling.""" storage_dir: Annotated[ str, diff --git a/src/crawlee/events/_local_event_manager.py b/src/crawlee/events/_local_event_manager.py index 2c94fcbe76..057d6f7f76 100644 --- a/src/crawlee/events/_local_event_manager.py +++ b/src/crawlee/events/_local_event_manager.py @@ -5,6 +5,8 @@ from logging import getLogger from typing import TYPE_CHECKING +import proclimits + from crawlee._utils.docs import docs_group from crawlee._utils.recurring_task import RecurringTask from crawlee._utils.system import get_cpu_info, get_memory_info @@ -48,6 +50,9 @@ def __init__( self._system_info_interval = system_info_interval """Interval between the emitted `SystemInfo` events.""" + self._cpu_load = proclimits.CpuLoad() + """CPU sampler of this event manager, measuring across the gap between its emissions.""" + self._emit_system_info_event_rec_task = RecurringTask( func=self._emit_system_info_event, delay=self._system_info_interval, @@ -76,6 +81,8 @@ async def __aenter__(self) -> Self: await super().__aenter__() if self._active_ref_count == 1: + # A reading kept from a previous session would report the average load over the idle gap since then. + self._cpu_load = proclimits.CpuLoad() self._emit_system_info_event_rec_task.start() return self @@ -98,10 +105,10 @@ async def __aexit__( async def _emit_system_info_event(self) -> None: """Emit a system info event with the current CPU and memory usage.""" - # Both readings block the thread they run in - `get_cpu_info` even samples the CPU utilization over a short + # Both readings block the thread they run in - `get_cpu_info` may sample the CPU utilization over a short # interval - so run them concurrently instead of one after the other. cpu_info, memory_info = await asyncio.gather( - asyncio.to_thread(get_cpu_info), + asyncio.to_thread(get_cpu_info, self._cpu_load), asyncio.to_thread(get_memory_info), ) diff --git a/tests/unit/_autoscaling/test_snapshotter.py b/tests/unit/_autoscaling/test_snapshotter.py index f86dea7ad2..068162cb8d 100644 --- a/tests/unit/_autoscaling/test_snapshotter.py +++ b/tests/unit/_autoscaling/test_snapshotter.py @@ -4,16 +4,18 @@ import time from bisect import insort from datetime import datetime, timedelta, timezone -from logging import getLogger +from logging import WARNING, getLogger from math import floor from typing import TYPE_CHECKING, Any, cast from unittest import mock -from unittest.mock import MagicMock +from unittest.mock import MagicMock, Mock +import proclimits import pytest from crawlee import service_locator from crawlee._autoscaling import Snapshotter +from crawlee._autoscaling import snapshotter as snapshotter_module from crawlee._autoscaling._types import ( SYSTEM_WIDE_MEMORY_OVERLOAD_THRESHOLD, ClientSnapshot, @@ -22,6 +24,7 @@ ) from crawlee._autoscaling.snapshotter import SortedSnapshotList from crawlee._utils.byte_size import ByteSize +from crawlee._utils.log import LoggerOnce from crawlee._utils.system import CpuInfo, MemoryInfo, get_memory_info from crawlee.configuration import Configuration from crawlee.events import LocalEventManager @@ -445,3 +448,33 @@ async def test_dynamic_memory( assert memory_samples[0].is_overloaded # Second sample can reflect the increased available memory based on the configuration used to create Snapshotter assert memory_samples[1].is_overloaded == (not dynamic_memory) + + +@pytest.mark.parametrize( + ('config_options', 'limit_bytes', 'expected_warning'), + [ + pytest.param({}, 512 * 1024**2, True, id='default ratio under a limit'), + pytest.param({'available_memory_ratio': 0.25}, 512 * 1024**2, False, id='ratio set explicitly'), + pytest.param({'memory_mbytes': 1024}, 512 * 1024**2, False, id='memory_mbytes set'), + pytest.param({}, None, False, id='no limit'), + ], +) +def test_from_config_warns_on_default_ratio_under_limit( + *, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + config_options: dict[str, Any], + limit_bytes: int | None, + expected_warning: bool, +) -> None: + """Warns when the default `available_memory_ratio` stacks on top of a memory limit.""" + budget = None if limit_bytes is None else proclimits.MemoryBudget(limit=limit_bytes, used=0, available=limit_bytes) + monkeypatch.setattr(proclimits, 'get_memory_budget', Mock(return_value=budget)) + monkeypatch.setattr(snapshotter_module, 'logger_once', LoggerOnce(snapshotter_module.logger)) + + with caplog.at_level(WARNING, logger=snapshotter_module.logger.name): + Snapshotter.from_config(Configuration(**config_options)) + + warnings = [record.message for record in caplog.records if 'Setting max memory of this run' in record.message] + assert len(warnings) == int(expected_warning) + assert all('to 25% of the 512.00 MB memory limit' in warning for warning in warnings) diff --git a/tests/unit/_utils/test_system.py b/tests/unit/_utils/test_system.py index a6634aa23e..277edc6e9e 100644 --- a/tests/unit/_utils/test_system.py +++ b/tests/unit/_utils/test_system.py @@ -8,6 +8,7 @@ from typing import TYPE_CHECKING from unittest.mock import Mock +import proclimits import psutil import pytest @@ -19,6 +20,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 +69,27 @@ 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 no real limit is measured.""" + sampler = Mock(spec=proclimits.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(proclimits, 'get_cpu_used_ratio', Mock(return_value=None)) + monkeypatch.setattr(proclimits, '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(proclimits, 'get_cpu_limit', Mock(return_value=1.0)) @pytest.fixture @@ -77,6 +99,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: proclimits.MemoryBudget | None = None, cpu_limit: float | None = None +) -> proclimits.Snapshot: + """Stand in for `proclimits.snapshot()`, describing an unrestricted process unless told otherwise.""" + return proclimits.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() @@ -206,11 +245,179 @@ def raise_error(*_args: object, **_kwargs: object) -> list[psutil.Process]: assert [record.getMessage() for record in caplog.records if 'child processes' in record.getMessage()] -def test_get_cpu_info_returns_valid_values() -> None: - cpu_info = get_cpu_info() +def test_get_cpu_info_returns_valid_values(cpu_load: Mock) -> None: + cpu_info = get_cpu_info(cpu_load) 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 = proclimits.MemoryBudget(limit=512 * 1024**2, used=100 * 1024**2, available=412 * 1024**2) + monkeypatch.setattr(proclimits, '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(proclimits, '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(cpu_load).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(proclimits, '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(cpu_load).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.usefixtures('_cpu_limited') +def test_get_cpu_info_measures_a_window_without_a_sampler(monkeypatch: pytest.MonkeyPatch) -> None: + """Without a sampler, the load is still measured against the limit.""" + get_cpu_used_ratio = Mock(return_value=0.25) + monkeypatch.setattr(proclimits, '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) + cpu_percent.assert_not_called() + + +@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(proclimits, '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(cpu_load).used_ratio == 0.0 + cpu_percent.assert_not_called() + + +@pytest.mark.usefixtures('_cpu_limited') +def test_get_cpu_info_falls_back_to_the_host_without_a_rate(monkeypatch: pytest.MonkeyPatch, cpu_load: Mock) -> None: + """A limit that no rate can be measured against is covered by the load of the host machine.""" + monkeypatch.setattr(psutil, 'cpu_percent', Mock(return_value=42.0)) + + assert get_cpu_info(cpu_load).used_ratio == 0.42 + + +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(proclimits, 'get_cpu_used_ratio', get_cpu_used_ratio) + monkeypatch.setattr(psutil, 'cpu_percent', Mock(return_value=42.0)) + + assert get_cpu_info(cpu_load).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( + proclimits.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: proclimits.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(proclimits, '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 any(expected_message in message for message in reported) + + +@pytest.mark.usefixtures('measured_current_process') +def test_log_resource_limits_reports_once(monkeypatch: pytest.MonkeyPatch) -> None: + """The limits are reported on the first sample only, not on every one.""" + snapshot = Mock(return_value=fake_snapshot()) + monkeypatch.setattr(proclimits, '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(proclimits, '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() + + # 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/tests/unit/events/test_local_event_manager.py b/tests/unit/events/test_local_event_manager.py index b88bba372d..323412dfbb 100644 --- a/tests/unit/events/test_local_event_manager.py +++ b/tests/unit/events/test_local_event_manager.py @@ -11,14 +11,15 @@ from crawlee.events._types import Event, EventSystemInfoData if TYPE_CHECKING: + import proclimits import pytest async def test_emit_system_info_event(monkeypatch: pytest.MonkeyPatch) -> None: """The recurring task emits the first `SystemInfo` event as soon as it starts, without waiting for the interval.""" - # Both readings are replaced with instant ones - a real `get_cpu_info` samples the CPU utilization over 100 ms, + # Both readings are replaced with instant ones - a real `get_cpu_info` may sample the CPU utilization over 100 ms, # and on a loaded runner it takes far longer than that. - monkeypatch.setattr('crawlee.events._local_event_manager.get_cpu_info', lambda: MagicMock(spec=CpuInfo)) + monkeypatch.setattr('crawlee.events._local_event_manager.get_cpu_info', lambda _: MagicMock(spec=CpuInfo)) monkeypatch.setattr('crawlee.events._local_event_manager.get_memory_info', lambda: MagicMock(spec=MemoryInfo)) mocked_listener = AsyncMock() @@ -43,7 +44,7 @@ async def test_system_info_readings_run_concurrently(monkeypatch: pytest.MonkeyP # A party left waiting alone breaks the barrier, which fails the test instead of hanging it. barrier = threading.Barrier(2, timeout=5) - def get_cpu_info_at_barrier() -> Any: + def get_cpu_info_at_barrier(_cpu_load: Any) -> Any: barrier.wait() return MagicMock(spec=CpuInfo) @@ -69,3 +70,30 @@ async def listener(event_data: EventSystemInfoData) -> None: await event_manager.wait_for_all_listeners_to_complete() assert len(received) == 1 + + +async def test_cpu_sampler_restarts_with_each_session(monkeypatch: pytest.MonkeyPatch) -> None: + """Each session samples the CPU afresh, instead of measuring across the idle gap since the previous one.""" + samplers: list[proclimits.CpuLoad] = [] + + def get_cpu_info(cpu_load: proclimits.CpuLoad) -> Any: + samplers.append(cpu_load) + return MagicMock(spec=CpuInfo) + + monkeypatch.setattr('crawlee.events._local_event_manager.get_cpu_info', get_cpu_info) + monkeypatch.setattr('crawlee.events._local_event_manager.get_memory_info', lambda: MagicMock(spec=MemoryInfo)) + + event_manager = LocalEventManager(system_info_interval=timedelta(hours=1)) + + for _ in range(2): + async with event_manager: + received = asyncio.Event() + + async def listener(_event_data: EventSystemInfoData, received: asyncio.Event = received) -> None: + received.set() + + event_manager.on(event=Event.SYSTEM_INFO, listener=listener) + await asyncio.wait_for(received.wait(), timeout=5) + + assert len(samplers) == 2 + assert samplers[0] is not samplers[1] diff --git a/uv.lock b/uv.lock index ee117b2be1..377d1041e7 100644 --- a/uv.lock +++ b/uv.lock @@ -3,10 +3,10 @@ revision = 3 requires-python = ">=3.10" resolution-markers = [ "python_full_version >= '3.14' and sys_platform == 'emscripten'", - "python_full_version == '3.13.*' and sys_platform == 'emscripten'", - "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version >= '3.14' and sys_platform != 'emscripten'", + "python_full_version == '3.13.*' and sys_platform == 'emscripten'", "python_full_version == '3.13.*' and sys_platform != 'emscripten'", + "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version == '3.12.*' and sys_platform != 'emscripten'", "python_full_version == '3.11.*'", "python_full_version < '3.11'", @@ -985,6 +985,7 @@ dependencies = [ { name = "colorama" }, { name = "impit" }, { name = "more-itertools" }, + { name = "proclimits" }, { name = "protego" }, { name = "psutil" }, { name = "pydantic" }, @@ -1170,6 +1171,7 @@ requires-dist = [ { name = "playwright", marker = "extra == 'adaptive-crawler'", specifier = ">=1.27.0" }, { name = "playwright", marker = "extra == 'playwright'", specifier = ">=1.27.0" }, { name = "playwright", marker = "extra == 'stagehand'", specifier = ">=1.27.0" }, + { name = "proclimits", specifier = ">=0.2.0" }, { name = "protego", specifier = ">=0.5.0" }, { name = "psutil", specifier = ">=6.0.0" }, { name = "pydantic", specifier = ">=2.11.0" }, @@ -2984,10 +2986,10 @@ version = "2.5.3" source = { registry = "https://pypi.org/simple" } resolution-markers = [ "python_full_version >= '3.14' and sys_platform == 'emscripten'", - "python_full_version == '3.13.*' and sys_platform == 'emscripten'", - "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version >= '3.14' and sys_platform != 'emscripten'", + "python_full_version == '3.13.*' and sys_platform == 'emscripten'", "python_full_version == '3.13.*' and sys_platform != 'emscripten'", + "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version == '3.12.*' and sys_platform != 'emscripten'", ] sdist = { url = "https://files.pythonhosted.org/packages/13/01/11703282db468b85f6f7b8c7f22d058de5970d5c7e60a3a8aaa313c3de36/numpy-2.5.3.tar.gz", hash = "sha256:df2d5874ff183595a4ba404edd04f6bd9b5505c1d7708573f6a6c17489a67563", size = 20791231, upload-time = "2026-09-06T16:27:47.073Z" } @@ -3357,6 +3359,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/45/e2/bbb7129c9e7999a6b8ee9cca3b66486c25c423ab5a75f34071798b74ce94/pre_commit-4.6.2-py2.py3-none-any.whl", hash = "sha256:e2dde9a75d3bce11bd3831c26d134df00a2803c1d818be6a0383c3dcda25dc4e", size = 226202, upload-time = "2026-08-10T22:07:16.942Z" }, ] +[[package]] +name = "proclimits" +version = "0.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/47/88/c96bbd2c272e4032c4e47374c8209f55a7e4414312feac3a8ca92121ae04/proclimits-0.2.0.tar.gz", hash = "sha256:fc80a1e6e993c63254836329ecc3f67555c8275fb4d5c3fd1c376b4aaab2d7b5", size = 51600, upload-time = "2026-09-17T14:28:05.612Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/14/b4/e2c68c699b2d1b3fedede8246216f7f71e8a4dd44112e1ee97910671ec7c/proclimits-0.2.0-py3-none-any.whl", hash = "sha256:3fd36ce31b5d72d9b4b06dc788c7c3257afec34f52a4dec5aa69b105d7cdcda7", size = 47690, upload-time = "2026-09-17T14:28:04.289Z" }, +] + [[package]] name = "propcache" version = "0.5.2" @@ -4339,10 +4350,10 @@ version = "1.9.1" source = { registry = "https://pypi.org/simple" } resolution-markers = [ "python_full_version >= '3.14' and sys_platform == 'emscripten'", - "python_full_version == '3.13.*' and sys_platform == 'emscripten'", - "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version >= '3.14' and sys_platform != 'emscripten'", + "python_full_version == '3.13.*' and sys_platform == 'emscripten'", "python_full_version == '3.13.*' and sys_platform != 'emscripten'", + "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version == '3.12.*' and sys_platform != 'emscripten'", "python_full_version == '3.11.*'", ] @@ -4540,10 +4551,10 @@ version = "1.18.1" source = { registry = "https://pypi.org/simple" } resolution-markers = [ "python_full_version >= '3.14' and sys_platform == 'emscripten'", - "python_full_version == '3.13.*' and sys_platform == 'emscripten'", - "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version >= '3.14' and sys_platform != 'emscripten'", + "python_full_version == '3.13.*' and sys_platform == 'emscripten'", "python_full_version == '3.13.*' and sys_platform != 'emscripten'", + "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version == '3.12.*' and sys_platform != 'emscripten'", ] dependencies = [ @@ -5460,10 +5471,10 @@ version = "17.1" source = { registry = "https://pypi.org/simple" } resolution-markers = [ "python_full_version >= '3.14' and sys_platform == 'emscripten'", - "python_full_version == '3.13.*' and sys_platform == 'emscripten'", - "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version >= '3.14' and sys_platform != 'emscripten'", + "python_full_version == '3.13.*' and sys_platform == 'emscripten'", "python_full_version == '3.13.*' and sys_platform != 'emscripten'", + "python_full_version == '3.12.*' and sys_platform == 'emscripten'", "python_full_version == '3.12.*' and sys_platform != 'emscripten'", "python_full_version == '3.11.*'", ]