Repository: apify/crawlee-python
File: src/crawlee/_autoscaling/snapshotter.py (Python)

Code before:

        This method does not perform memory usage measurement. Instead, it just reads the data received through
        the `event_data` parameter, which is expected to be supplied by the event manager.

        Args:
            event_data: System info data from which memory usage is read.
        """
        snapshot = MemorySnapshot(
            current_size=event_data.memory_info.current_size,
            max_memory_size=self._max_memory_size,
            max_used_memory_ratio=self._max_used_memory_ratio,
            created_at=event_data.memory_info.created_at,
            system_wide_used_size=None,
            system_wide_memory_size=None,
        )

        if isinstance(memory_info := event_data.memory_info, MemoryInfo):
            snapshot.system_wide_used_size = memory_info.system_wide_used_size
            snapshot.system_wide_memory_size = memory_info.total_size

        snapshots = cast('list[Snapshot]', self._memory_snapshots)
        self._prune_snapshots(snapshots, snapshot.created_at)
        self._memory_snapshots.add(snapshot)

        if isinstance(self._max_memory_size, Ratio):
            max_memory_size = ByteSize(int(get_memory_info().total_size.bytes * self._max_memory_size.value))
        else:
            max_memory_size = self._max_memory_size

        self._evaluate_memory_load(
            event_data.memory_info.current_size,
            event_data.memory_info.created_at,
            max_memory_size=max_memory_size,
        )

    def _snapshot_event_loop(self) -> None:
        """Capture a snapshot of the current event loop usage.

        This method evaluates the event loop's latency by comparing the expected time between snapshots to the actual
        time elapsed since the last snapshot. The delay in the snapshot reflects the time deviation due to event loop
        overhead - it's calculated by subtracting the expected interval between snapshots from the actual time elapsed
        since the last snapshot. If there's no previous snapshot, the delay is considered zero.
        """
        snapshot = EventLoopSnapshot(max_delay=self._max_event_loop_delay, delay=timedelta(seconds=0))
        previous_snapshot = self._event_loop_snapshots[-1] if self._event_loop_snapshots else None

        if previous_snapshot:
            event_loop_delay = snapshot.created_at - previous_snapshot.created_at - self._EVENT_LOOP_SNAPSHOT_INTERVAL
            snapshot.delay = event_loop_delay

        snapshots = cast('list[Snapshot]', self._event_loop_snapshots)

Review comment (suggestion):
I think we could skip calling `get_memory_info()` when `event_data.memory_info` is `MemoryInfo` and just use `memory_info.total_size.bytes` instead.

Code after:

        snapshots = cast('list[Snapshot]', self._cpu_snapshots)
        self._prune_snapshots(snapshots, event_data.cpu_info.created_at)
        self._cpu_snapshots.add(snapshot)

    def _snapshot_memory(self, event_data: EventSystemInfoData) -> None:
        """Capture a snapshot of the current memory usage.

        This method does not perform memory usage measurement. Instead, it just reads the data received through
        the `event_data` parameter, which is expected to be supplied by the event manager.

        Args:
            event_data: System info data from which memory usage is read.
        """
        match event_data.memory_info, self._max_memory_size:
            case MemoryInfo() as memory_info, Ratio() as ratio:
                max_memory_size = memory_info.total_size * ratio.value
                system_wide_used_size = memory_info.system_wide_used_size
                system_wide_memory_size = memory_info.total_size

            case MemoryUsageInfo(), Ratio() as ratio:
                # This is just hypothetical case, that will most likely not happen in practice.
                # `LocalEventManager` should always provide `MemoryInfo` in the event data.
                # When running on Apify, `self._max_memory_size` is always `ByteSize`, not `Ratio`.
                _warn_once(
                    'It is recommended that a custom implementation of `LocalEventManager` emits `SYSTEM_INFO` events '
                    'with `MemoryInfo` and not just `MemoryUsageInfo`.'
                )
                max_memory_size = get_memory_info().total_size * ratio.value
                system_wide_used_size = None
                system_wide_memory_size = None

            case MemoryInfo() as memory_info, ByteSize() as byte_size:
                max_memory_size = byte_size
                system_wide_used_size = memory_info.system_wide_used_size
                system_wide_memory_size = memory_info.total_size

            case MemoryUsageInfo(), ByteSize() as byte_size:
                max_memory_size = byte_size
                system_wide_used_size = None
                system_wide_memory_size = None

            case _, _:
                raise NotImplementedError('Unsupported combination of memory info and max memory size types.')

        snapshot = MemorySnapshot(
            current_size=event_data.memory_info.current_size,
            max_memory_size=max_memory_size,
            max_used_memory_ratio=self._max_used_memory_ratio,
            created_at=event_data.memory_info.created_at,
            system_wide_used_size=system_wide_used_size,