fix: Fix snapshots handling (#692)

Small fixes to snapshot handling:
- Ensure time ordering of snapshots.
- Remove fake time weight for samples from same time.
- Remove unnecessary reversing of snapshot.
- Add test for memory and cpu snapshot ordering.

---------

Co-authored-by: Vlada Dusek <[email protected]>

Author: Pijukatel

poetry.lock

@@ -86,6 +86,7 @@ pytest-xdist = "~3.6.0"
 respx = "~0.21.0"
 ruff = "~0.7.0"
 setuptools = "~75.3.0"                    # setuptools are used by pytest, but not explicitly required
+sortedcontainers-stubs = "^2.4.2"
 types-beautifulsoup4 = "~4.12.0.20240229"
 types-colorama = "~0.4.15.20240106"
 types-psutil = "~5.9.5.20240205"

pyproject.toml

@@ -4,11 +4,19 @@
 
 from datetime import datetime, timedelta, timezone
 from logging import getLogger
-from typing import TYPE_CHECKING, cast
+from operator import attrgetter
+from typing import TYPE_CHECKING, TypeVar, cast
 
 import psutil
-
-from crawlee._autoscaling.types import ClientSnapshot, CpuSnapshot, EventLoopSnapshot, MemorySnapshot, Snapshot
+from sortedcontainers import SortedList
+
+from crawlee._autoscaling.types import (
+    ClientSnapshot,
+    CpuSnapshot,
+    EventLoopSnapshot,
+    MemorySnapshot,
+    Snapshot,
+)
 from crawlee._utils.byte_size import ByteSize
 from crawlee._utils.docs import docs_group
 from crawlee._utils.recurring_task import RecurringTask
@@ -21,6 +29,8 @@
 
 logger = getLogger(__name__)
 
+T = TypeVar('T')
+
 
 @docs_group('Classes')
 class Snapshotter:
@@ -94,16 +104,20 @@ def __init__(
             cast(float, available_memory_ratio)
         )
 
-        self._cpu_snapshots: list[CpuSnapshot] = []
-        self._event_loop_snapshots: list[EventLoopSnapshot] = []
-        self._memory_snapshots: list[MemorySnapshot] = []
-        self._client_snapshots: list[ClientSnapshot] = []
+        self._cpu_snapshots = self._get_sorted_list_by_created_at(list[CpuSnapshot]())
+        self._event_loop_snapshots = self._get_sorted_list_by_created_at(list[EventLoopSnapshot]())
+        self._memory_snapshots = self._get_sorted_list_by_created_at(list[MemorySnapshot]())
+        self._client_snapshots = self._get_sorted_list_by_created_at(list[ClientSnapshot]())
 
         self._snapshot_event_loop_task = RecurringTask(self._snapshot_event_loop, self._event_loop_snapshot_interval)
         self._snapshot_client_task = RecurringTask(self._snapshot_client, self._client_snapshot_interval)
 
         self._timestamp_of_last_memory_warning: datetime = datetime.now(timezone.utc) - timedelta(hours=1)
 
+    @staticmethod
+    def _get_sorted_list_by_created_at(input_list: list[T]) -> SortedList[T]:
+        return SortedList(input_list, key=attrgetter('created_at'))
+
     @staticmethod
     def _get_default_max_memory_size(available_memory_ratio: float) -> ByteSize:
         """Default `memory_max_size` is 1/4 of the total system memory."""
@@ -194,7 +208,7 @@ def _get_sample(snapshots: list[Snapshot], duration: timedelta | None = None) ->
             return []
 
         latest_time = snapshots[-1].created_at
-        return [snapshot for snapshot in reversed(snapshots) if latest_time - snapshot.created_at <= duration]
+        return [snapshot for snapshot in snapshots if latest_time - snapshot.created_at <= duration]
 
     def _snapshot_cpu(self, event_data: EventSystemInfoData) -> None:
         """Captures a snapshot of the current CPU usage.
@@ -213,7 +227,7 @@ def _snapshot_cpu(self, event_data: EventSystemInfoData) -> None:
 
         snapshots = cast(list[Snapshot], self._cpu_snapshots)
         self._prune_snapshots(snapshots, event_data.cpu_info.created_at)
-        self._cpu_snapshots.append(snapshot)
+        self._cpu_snapshots.add(snapshot)
 
     def _snapshot_memory(self, event_data: EventSystemInfoData) -> None:
         """Captures a snapshot of the current memory usage.
@@ -233,8 +247,7 @@ def _snapshot_memory(self, event_data: EventSystemInfoData) -> None:
 
         snapshots = cast(list[Snapshot], self._memory_snapshots)
         self._prune_snapshots(snapshots, snapshot.created_at)
-        self._memory_snapshots.append(snapshot)
-
+        self._memory_snapshots.add(snapshot)
         self._evaluate_memory_load(event_data.memory_info.current_size, event_data.memory_info.created_at)
 
     def _snapshot_event_loop(self) -> None:
@@ -254,7 +267,7 @@ def _snapshot_event_loop(self) -> None:
 
         snapshots = cast(list[Snapshot], self._event_loop_snapshots)
         self._prune_snapshots(snapshots, snapshot.created_at)
-        self._event_loop_snapshots.append(snapshot)
+        self._event_loop_snapshots.add(snapshot)
 
     def _snapshot_client(self) -> None:
         """Captures a snapshot of the current API state by checking for rate limit errors (HTTP 429).
@@ -271,7 +284,7 @@ def _snapshot_client(self) -> None:
 
         snapshots = cast(list[Snapshot], self._client_snapshots)
         self._prune_snapshots(snapshots, snapshot.created_at)
-        self._client_snapshots.append(snapshot)
+        self._client_snapshots.add(snapshot)
 
     def _prune_snapshots(self, snapshots: list[Snapshot], now: datetime) -> None:
         """Removes snapshots that are older than the `self._snapshot_history`.