diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md new file mode 100644 index 0000000000..07bdbf5ad0 --- /dev/null +++ b/docs/upgrading/upgrading_to_v2.md @@ -0,0 +1,15 @@ +--- +id: upgrading-to-v2 +title: Upgrading to v2 +--- + +This page summarizes the breaking changes between Crawlee for Python v1.x and v2.0. + +## Statistics + +The runtime tracking in `StatisticsState` was reworked to fix `crawler_runtime` miscounting after migrations and resurrections. The runtime accumulated by previous runs is now restored from the persisted `crawlerRuntimeMillis` value, so it survives any number of interruptions, and the downtime between the runs is no longer counted. + +- `StatisticsState.crawler_runtime` is a read-only property now. The deprecated setter was removed; assigning to it raises `AttributeError` instead of emitting a `DeprecationWarning`. +- `StatisticsState.crawler_runtime_for_serialization` was removed. The persisted state still contains `crawlerRuntimeMillis`, now written by the `runtime_offset` field, which also restores the value when a state is loaded. +- `crawlerRuntimeMillis` is serialized as a number of milliseconds, consistent with the other `*Millis` fields and with Crawlee for JavaScript, instead of an ISO 8601 duration string. States persisted by v1.x load correctly. +- The unused state fields `errors`, `retry_errors`, `requests_finished_per_minute` and `requests_failed_per_minute` were removed from `StatisticsState`. They were never populated. The `FinalStatistics` fields of the same names are unaffected, and the per-minute rates in the statistics logs are still computed. diff --git a/src/crawlee/statistics/_models.py b/src/crawlee/statistics/_models.py index 428332f23e..04eb3c4086 100644 --- a/src/crawlee/statistics/_models.py +++ b/src/crawlee/statistics/_models.py @@ -1,12 +1,22 @@ from __future__ import annotations import json -import warnings from dataclasses import asdict, dataclass from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Annotated, Any -from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, PlainValidator, computed_field +from pydantic import ( + BaseModel, + ConfigDict, + Field, + PlainSerializer, + PlainValidator, + ValidationError, + ValidatorFunctionWrapHandler, + WrapValidator, + computed_field, + field_serializer, +) from typing_extensions import override from crawlee._utils.console import make_table @@ -54,6 +64,14 @@ def __str__(self) -> str: ) +def _runtime_offset_or_none(value: Any, handler: ValidatorFunctionWrapHandler) -> timedelta | None: + """Treat an invalid persisted runtime value as absent, so that it does not prevent loading the whole state.""" + try: + return handler(value) + except ValidationError: + return None + + @docs_group('Statistics') class StatisticsState(BaseModel): """Statistic data about a crawler run.""" @@ -64,8 +82,6 @@ class StatisticsState(BaseModel): requests_finished: Annotated[int, Field(alias='requestsFinished')] = 0 requests_failed: Annotated[int, Field(alias='requestsFailed')] = 0 requests_retries: Annotated[int, Field(alias='requestsRetries')] = 0 - requests_failed_per_minute: Annotated[float, Field(alias='requestsFailedPerMinute')] = 0 - requests_finished_per_minute: Annotated[float, Field(alias='requestsFinishedPerMinute')] = 0 request_min_duration: Annotated[timedelta_ms | None, Field(alias='requestMinDurationMillis')] = None request_max_duration: Annotated[timedelta_ms | None, Field(alias='requestMaxDurationMillis')] = None request_total_failed_duration: Annotated[timedelta_ms, Field(alias='requestTotalFailedDurationMillis')] = ( @@ -80,12 +96,8 @@ class StatisticsState(BaseModel): # Workaround for Pydantic and type checkers when using Annotated with default_factory if TYPE_CHECKING: - errors: dict[str, Any] = {} - retry_errors: dict[str, Any] = {} requests_with_status_code: dict[str, int] = {} else: - errors: Annotated[dict[str, Any], Field(default_factory=dict)] - retry_errors: Annotated[dict[str, Any], Field(alias='retryErrors', default_factory=dict)] requests_with_status_code: Annotated[ dict[str, int], Field(alias='requestsWithStatusCode', default_factory=dict), @@ -104,36 +116,39 @@ class StatisticsState(BaseModel): ), ] = {} - # Used to track the crawler runtime, that had already been persisted. This is the runtime from previous runs. - _runtime_offset: Annotated[timedelta, Field(exclude=True)] = timedelta() + # The runtime accumulated by previous runs, restored from the persisted `crawlerRuntimeMillis` value. + # When serialized, the live `crawler_runtime` is written instead, so that a persisted state always + # carries the total runtime accumulated so far. + runtime_offset: Annotated[ + timedelta_ms | None, + WrapValidator(_runtime_offset_or_none), + Field(alias='crawlerRuntimeMillis'), + ] = None def model_post_init(self, /, __context: Any) -> None: - self._runtime_offset = self.crawler_runtime or self._runtime_offset + # When the runtime accumulated by previous runs is not available (a state persisted by an older + # version, or an invalid persisted value), reconstruct it from the timestamps. The end of a run + # that did not finish cleanly (migration, abort) is approximated by the moment the state was last + # persisted, so that the downtime before this run is not counted towards the runtime. + if self.runtime_offset is None: + if self.crawler_last_started_at: + finished_at = self.crawler_finished_at or self.stats_persisted_at or datetime.now(timezone.utc) + self.runtime_offset = finished_at - self.crawler_last_started_at + else: + self.runtime_offset = timedelta() + self.runtime_offset = max(timedelta(), self.runtime_offset) @property def crawler_runtime(self) -> timedelta: + offset = self.runtime_offset or timedelta() if self.crawler_last_started_at: finished_at = self.crawler_finished_at or datetime.now(timezone.utc) - return self._runtime_offset + finished_at - self.crawler_last_started_at - return self._runtime_offset - - @crawler_runtime.setter - def crawler_runtime(self, value: timedelta) -> None: - # Setter for backwards compatibility only, the crawler_runtime is now computed_field, and can't be set manually. - # To be removed in v2 release https://github.com/apify/crawlee-python/issues/1567 - warnings.warn( - f"Setting 'crawler_runtime' is deprecated and will be removed in a future version." - f' Value {value} will not be used.', - DeprecationWarning, - stacklevel=2, - ) + return offset + finished_at - self.crawler_last_started_at + return offset - @computed_field(alias='crawlerRuntimeMillis') - def crawler_runtime_for_serialization(self) -> timedelta: - if self.crawler_last_started_at: - finished_at = self.crawler_finished_at or datetime.now(timezone.utc) - return self._runtime_offset + finished_at - self.crawler_last_started_at - return self._runtime_offset + @field_serializer('runtime_offset') + def _serialize_runtime_offset(self, _value: timedelta | None) -> float: + return round(self.crawler_runtime.total_seconds() * 1000) @computed_field(alias='requestTotalDurationMillis', return_type=timedelta_ms) @property diff --git a/src/crawlee/statistics/_statistics.py b/src/crawlee/statistics/_statistics.py index 3d568fdda2..91d76a9eae 100644 --- a/src/crawlee/statistics/_statistics.py +++ b/src/crawlee/statistics/_statistics.py @@ -165,7 +165,12 @@ async def __aenter__(self) -> Self: raise RuntimeError(f'The {self.__class__.__name__} is already active.') await self._state.initialize() - # Reset `crawler_finished_at` to indicate a new run in progress. + + # The runtime accumulated by previous runs is restored into the state's `runtime_offset`. Reset the + # timestamps so that the initial periodic log prints only that accumulated runtime (zero for a fresh + # start), instead of measuring against the previous run's start time, which would include the + # downtime between the runs (e.g. after a migration or resurrection). + self.state.crawler_last_started_at = None self.state.crawler_finished_at = None # Start periodic logging and let it print initial state before activation. diff --git a/tests/unit/_statistics/test_persistence.py b/tests/unit/_statistics/test_persistence.py index 3c6e06d02d..1c26466ed6 100644 --- a/tests/unit/_statistics/test_persistence.py +++ b/tests/unit/_statistics/test_persistence.py @@ -1,6 +1,12 @@ from __future__ import annotations +import logging +from datetime import datetime, timedelta, timezone + +import pytest + from crawlee.statistics import Statistics +from crawlee.storages import KeyValueStore async def test_basic_persistence() -> None: @@ -13,3 +19,179 @@ async def test_basic_persistence() -> None: pass assert statistics.state.requests_failed == 42 + + +async def test_first_periodic_log_of_fresh_run_reports_zero_runtime(caplog: pytest.LogCaptureFixture) -> None: + """The first periodic log of a fresh run must report a runtime of exactly zero.""" + caplog.set_level(logging.INFO) + log_message = 'Fresh statistics' + + async with Statistics.with_default_state(log_message=log_message, statistics_log_format='inline'): + pass + + periodic_records = [record for record in caplog.records if record.message == log_message] + assert periodic_records + assert periodic_records[0].crawler_runtime == 0 # ty: ignore[unresolved-attribute] + + +async def test_periodic_log_after_resume_excludes_downtime(caplog: pytest.LogCaptureFixture) -> None: + """The first periodic log of a resumed run must report only the previous runtime, without the downtime.""" + caplog.set_level(logging.INFO) + key = 'statistics_downtime_clean' + log_message = 'Statistics after resume' + downtime = timedelta(hours=2) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key): + pass + + # Simulate a resurrection after two hours of downtime by shifting the persisted timestamps into the past. + kvs = await KeyValueStore.open() + stored_state = await kvs.get_value(key) + for field in ('crawlerStartedAt', 'crawlerLastStartTimestamp', 'crawlerFinishedAt', 'statsPersistedAt'): + # `datetime.fromisoformat` does not accept the 'Z' suffix until Python 3.11. + stored_timestamp = datetime.fromisoformat(stored_state[field].replace('Z', '+00:00')) + stored_state[field] = (stored_timestamp - downtime).isoformat() + await kvs.set_value(key, stored_state) + + caplog.clear() + async with Statistics.with_default_state( + persistence_enabled=True, + persist_state_key=key, + log_message=log_message, + statistics_log_format='inline', + ): + pass + + periodic_records = [record for record in caplog.records if record.message == log_message] + assert periodic_records + first_logged_runtime = timedelta(seconds=periodic_records[0].crawler_runtime) # ty: ignore[unresolved-attribute] + previous_runtime = timedelta(milliseconds=stored_state['crawlerRuntimeMillis']) + assert abs(first_logged_runtime - previous_runtime) < timedelta(milliseconds=1) + + +async def test_runtime_accumulates_over_multiple_resurrections() -> None: + """The persisted total runtime is restored as the runtime offset, so runs before the last one still count.""" + key = 'statistics_accumulated_runtime' + now = datetime.now(timezone.utc) + last_run_start = now - timedelta(hours=1) + last_run_duration = timedelta(seconds=5) + total_runtime = timedelta(seconds=30) + + # State persisted after a clean run whose own segment took 5s, with 30s of runtime accumulated in total. + kvs = await KeyValueStore.open() + await kvs.set_value( + key, + { + 'crawlerStartedAt': (now - timedelta(hours=2)).isoformat(), + 'crawlerLastStartTimestamp': last_run_start.isoformat(), + 'crawlerFinishedAt': (last_run_start + last_run_duration).isoformat(), + 'statsPersistedAt': (last_run_start + last_run_duration).isoformat(), + 'crawlerRuntimeMillis': total_runtime.total_seconds() * 1000, + }, + ) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics: + runtime = statistics.state.crawler_runtime + + assert total_runtime <= runtime < total_runtime + timedelta(minutes=1) + + +async def test_v1_persisted_runtime_is_restored() -> None: + """States persisted by v1.x store `crawlerRuntimeMillis` as an ISO 8601 duration string - it must load.""" + key = 'statistics_v1_runtime' + now = datetime.now(timezone.utc) + total_runtime = timedelta(seconds=30) + + kvs = await KeyValueStore.open() + await kvs.set_value( + key, + { + 'crawlerStartedAt': (now - timedelta(hours=2)).isoformat(), + 'crawlerLastStartTimestamp': (now - timedelta(hours=1)).isoformat(), + 'crawlerFinishedAt': (now - timedelta(hours=1) + timedelta(seconds=5)).isoformat(), + 'statsPersistedAt': (now - timedelta(hours=1) + timedelta(seconds=5)).isoformat(), + 'crawlerRuntimeMillis': 'PT30S', + }, + ) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics: + runtime = statistics.state.crawler_runtime + + assert total_runtime <= runtime < total_runtime + timedelta(minutes=1) + + +@pytest.mark.parametrize('invalid_runtime', [None, 'N/A'], ids=['null', 'garbage']) +async def test_invalid_persisted_runtime_falls_back_to_timestamps(invalid_runtime: str | None) -> None: + """An invalid persisted runtime value must not prevent loading the state. + + The runtime is approximated from the timestamps instead, as for states persisted by older versions.""" + key = 'statistics_invalid_runtime' + now = datetime.now(timezone.utc) + downtime = timedelta(hours=2) + previous_runtime = timedelta(seconds=10) + + kvs = await KeyValueStore.open() + await kvs.set_value( + key, + { + 'crawlerStartedAt': (now - downtime - previous_runtime).isoformat(), + 'crawlerLastStartTimestamp': (now - downtime - previous_runtime).isoformat(), + 'crawlerFinishedAt': None, + 'statsPersistedAt': (now - downtime).isoformat(), + 'crawlerRuntimeMillis': invalid_runtime, + }, + ) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics: + runtime = statistics.state.crawler_runtime + + assert previous_runtime <= runtime < previous_runtime + downtime + + +async def test_negative_persisted_runtime_is_clamped_to_zero() -> None: + """A negative persisted runtime value (e.g. after a backwards clock jump) must not skew the statistics.""" + key = 'statistics_negative_runtime' + now = datetime.now(timezone.utc) + + kvs = await KeyValueStore.open() + await kvs.set_value( + key, + { + 'crawlerStartedAt': (now - timedelta(hours=1)).isoformat(), + 'crawlerLastStartTimestamp': (now - timedelta(hours=1)).isoformat(), + 'crawlerFinishedAt': (now - timedelta(hours=1)).isoformat(), + 'statsPersistedAt': (now - timedelta(hours=1)).isoformat(), + 'crawlerRuntimeMillis': -5000, + }, + ) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics: + runtime = statistics.state.crawler_runtime + + assert timedelta() <= runtime < timedelta(minutes=1) + + +async def test_runtime_after_unclean_shutdown_excludes_downtime() -> None: + """State persisted mid-run (migration, abort): the runtime of the previous run is approximated by the moment + the state was last persisted, so the downtime before the resumed run must not inflate the runtime.""" + key = 'statistics_downtime_unclean' + now = datetime.now(timezone.utc) + downtime = timedelta(hours=2) + previous_runtime = timedelta(seconds=10) + + kvs = await KeyValueStore.open() + await kvs.set_value( + key, + { + 'requestsFinished': 2, + 'crawlerStartedAt': (now - downtime - previous_runtime).isoformat(), + 'crawlerLastStartTimestamp': (now - downtime - previous_runtime).isoformat(), + 'crawlerFinishedAt': None, + 'statsPersistedAt': (now - downtime).isoformat(), + }, + ) + + async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics: + runtime = statistics.state.crawler_runtime + + assert previous_runtime <= runtime < previous_runtime + downtime