Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/guides/scaling_crawlers.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -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 automatically starts new 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 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 <ApiLink to="class/Configuration">`Configuration`</ApiLink>, together with `memory_mbytes` for sizing the budget in absolute terms.
5 changes: 5 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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 <task>
[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 "
Expand Down
90 changes: 81 additions & 9 deletions src/crawlee/_utils/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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.
Expand Down Expand Up @@ -185,35 +193,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, 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
Expand All @@ -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
Loading