respect the memory and CPU limits that apply to the process

Author: Mantisus

docs/guides/scaling_crawlers.mdx

@@ -47,3 +47,9 @@ The `desired_concurrency` option in the <ApiLink to="class/ConcurrencySettings">
 ## Autoscaled pool
 
 The <ApiLink to="class/AutoscaledPool">`AutoscaledPool`</ApiLink> 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 <ApiLink to="class/Snapshotter">`Snapshotter`</ApiLink> and <ApiLink to="class/SystemStatus">`SystemStatus`</ApiLink> classes. If any task raises an exception, the error is propagated, and the pool is stopped. Every crawler uses an <ApiLink to="class/AutoscaledPool">`AutoscaledPool`</ApiLink> 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 <ApiLink to="class/Configuration">`Configuration`</ApiLink>, together with `memory_mbytes` for sizing the budget in absolute terms. 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.

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",

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`."""

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,35 +189,83 @@ 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,
         PlainValidator(ByteSize.validate),
         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
 
+    if limits.memory_budget is None and cores is None:
+        return
 
-def get_cpu_info() -> CpuInfo:
+    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) -> 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.
     """
     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 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 +288,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),
     )

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,

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),
         )