feat: Add StatusMessageWatcher (#407)

### Description

Add the option to log status and status messages of another Actor run.
If such Actor run is also redirecting logs from another Actor run, then
those logs can propagate all the way to the top through the
`StreamedLog` - deep redirection. If the Actor run from which we are
redirecting status messages is not redirecting logs from its children,
then the log and status redirection remains shallow - only from that
Actor run.

### Issues

- Closes: #404

Author: Pijukatel

src/apify_client/_logging.py

@@ -164,5 +164,5 @@ def format(self, record: logging.LogRecord) -> str:
         Returns:
             Formated log message.
         """
-        formated_logger_name = f'{Fore.CYAN}[{record.name}]{Style.RESET_ALL} '
-        return f'{formated_logger_name}-> {record.msg}'
+        formated_logger_name = f'{Fore.CYAN}[{record.name}]{Style.RESET_ALL}'
+        return f'{formated_logger_name} -> {record.msg}'

src/apify_client/clients/resource_clients/actor.py

@@ -317,7 +317,8 @@ def call(
                 waits indefinitely.
             logger: Logger used to redirect logs from the Actor run. Using "default" literal means that a predefined
                 default logger will be used. Setting `None` will disable any log propagation. Passing custom logger
-                will redirect logs to the provided logger.
+                will redirect logs to the provided logger. The logger is also used to capture status and status message
+                of the other Actor run.
 
         Returns:
             The run object.
@@ -336,12 +337,11 @@ def call(
             return self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
         run_client = self.root_client.run(run_id=started_run['id'])
+
         if logger == 'default':
-            log_context = run_client.get_streamed_log()
-        else:
-            log_context = run_client.get_streamed_log(to_logger=logger)
+            logger = None
 
-        with log_context:
+        with run_client.get_status_message_watcher(to_logger=logger), run_client.get_streamed_log(to_logger=logger):
             return self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
     def build(
@@ -722,7 +722,8 @@ async def call(
                 waits indefinitely.
             logger: Logger used to redirect logs from the Actor run. Using "default" literal means that a predefined
                 default logger will be used. Setting `None` will disable any log propagation. Passing custom logger
-                will redirect logs to the provided logger.
+                will redirect logs to the provided logger. The logger is also used to capture status and status message
+                of the other Actor run.
 
         Returns:
             The run object.
@@ -742,12 +743,14 @@ async def call(
             return await self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
         run_client = self.root_client.run(run_id=started_run['id'])
+
         if logger == 'default':
-            log_context = await run_client.get_streamed_log()
-        else:
-            log_context = await run_client.get_streamed_log(to_logger=logger)
+            logger = None
+
+        status_redirector = await run_client.get_status_message_watcher(to_logger=logger)
+        streamed_log = await run_client.get_streamed_log(to_logger=logger)
 
-        async with log_context:
+        async with status_redirector, streamed_log:
             return await self.root_client.run(started_run['id']).wait_for_finish(wait_secs=wait_secs)
 
     async def build(

src/apify_client/clients/resource_clients/log.py

@@ -4,9 +4,10 @@
 import logging
 import re
 import threading
+import time
 from asyncio import Task
 from contextlib import asynccontextmanager, contextmanager
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from threading import Thread
 from typing import TYPE_CHECKING, Any, cast
 
@@ -23,6 +24,8 @@
     import httpx
     from typing_extensions import Self
 
+    from apify_client.clients import RunClient, RunClientAsync
+
 
 class LogClient(ResourceClient):
     """Sub-client for manipulating logs."""
@@ -228,9 +231,9 @@ def __init__(self, to_logger: logging.Logger, *, from_start: bool = True) -> Non
                 logs for long-running actors in stand-by.
 
         """
-        self._to_logger = to_logger
         if self._force_propagate:
             to_logger.propagate = True
+        self._to_logger = to_logger
         self._stream_buffer = list[bytes]()
         self._split_marker = re.compile(rb'(?:\n|^)(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z)')
         self._relevancy_time_limit: datetime | None = None if from_start else datetime.now(tz=timezone.utc)
@@ -350,13 +353,16 @@ def start(self) -> Task:
         self._streaming_task = asyncio.create_task(self._stream_log())
         return self._streaming_task
 
-    def stop(self) -> None:
+    async def stop(self) -> None:
         """Stop the streaming task."""
         if not self._streaming_task:
             raise RuntimeError('Streaming task is not active')
 
         self._streaming_task.cancel()
-        self._streaming_task = None
+        try:
+            await self._streaming_task
+        except asyncio.CancelledError:
+            self._streaming_task = None
 
     async def __aenter__(self) -> Self:
         """Start the streaming task within the context. Exiting the context will cancel the streaming task."""
@@ -367,7 +373,7 @@ async def __aexit__(
         self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
     ) -> None:
         """Cancel the streaming task."""
-        self.stop()
+        await self.stop()
 
     async def _stream_log(self) -> None:
         async with self._log_client.stream(raw=True) as log_stream:
@@ -378,3 +384,163 @@ async def _stream_log(self) -> None:
 
             # If the stream is finished, then the last part will be also processed.
             self._log_buffer_content(include_last_part=True)
+
+
+class StatusMessageWatcher:
+    """Utility class for logging status messages from another Actor run.
+
+    Status message is logged at fixed time intervals, and there is no guarantee that all messages will be logged,
+    especially in cases of frequent status message changes.
+    """
+
+    _force_propagate = False
+    # This is final sleep time to try to get the last status and status message of finished Actor run.
+    # The status and status message can get set on the Actor run with a delay. Sleep time does not guarantee that the
+    # final message will be captured, but increases the chances of that.
+    _final_sleep_time_s = 6
+
+    def __init__(self, *, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=5)) -> None:
+        """Initialize `StatusMessageWatcher`.
+
+        Args:
+            to_logger: The logger to which the status message will be redirected.
+            check_period: The period with which the status message will be polled.
+        """
+        if self._force_propagate:
+            to_logger.propagate = True
+        self._to_logger = to_logger
+        self._check_period = check_period.total_seconds()
+        self._last_status_message = ''
+
+    def _log_run_data(self, run_data: dict[str, Any] | None) -> bool:
+        """Get relevant run data, log them if changed and return `True` if more data is expected.
+
+        Args:
+            run_data: The dictionary that contains the run data.
+
+        Returns:
+              `True` if more data is expected, `False` otherwise.
+        """
+        if run_data is not None:
+            status = run_data.get('status', 'Unknown status')
+            status_message = run_data.get('statusMessage', '')
+            new_status_message = f'Status: {status}, Message: {status_message}'
+
+            if new_status_message != self._last_status_message:
+                self._last_status_message = new_status_message
+                self._to_logger.info(new_status_message)
+
+            return not (run_data.get('isStatusMessageTerminal', False))
+        return True
+
+
+class StatusMessageWatcherAsync(StatusMessageWatcher):
+    """Async variant of `StatusMessageWatcher` that is logging in task."""
+
+    def __init__(
+        self, *, run_client: RunClientAsync, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=1)
+    ) -> None:
+        """Initialize `StatusMessageWatcherAsync`.
+
+        Args:
+            run_client: The client for run that will be used to get a status and message.
+            to_logger: The logger to which the status message will be redirected.
+            check_period: The period with which the status message will be polled.
+        """
+        super().__init__(to_logger=to_logger, check_period=check_period)
+        self._run_client = run_client
+        self._logging_task: Task | None = None
+
+    def start(self) -> Task:
+        """Start the logging task. The caller has to handle any cleanup by manually calling the `stop` method."""
+        if self._logging_task:
+            raise RuntimeError('Logging task already active')
+        self._logging_task = asyncio.create_task(self._log_changed_status_message())
+        return self._logging_task
+
+    async def stop(self) -> None:
+        """Stop the logging task."""
+        if not self._logging_task:
+            raise RuntimeError('Logging task is not active')
+
+        self._logging_task.cancel()
+        try:
+            await self._logging_task
+        except asyncio.CancelledError:
+            self._logging_task = None
+
+    async def __aenter__(self) -> Self:
+        """Start the logging task within the context. Exiting the context will cancel the logging task."""
+        self.start()
+        return self
+
+    async def __aexit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> None:
+        """Cancel the logging task."""
+        await asyncio.sleep(self._final_sleep_time_s)
+        await self.stop()
+
+    async def _log_changed_status_message(self) -> None:
+        while True:
+            run_data = await self._run_client.get()
+            if not self._log_run_data(run_data):
+                break
+            await asyncio.sleep(self._check_period)
+
+
+class StatusMessageWatcherSync(StatusMessageWatcher):
+    """Sync variant of `StatusMessageWatcher` that is logging in thread."""
+
+    def __init__(
+        self, *, run_client: RunClient, to_logger: logging.Logger, check_period: timedelta = timedelta(seconds=1)
+    ) -> None:
+        """Initialize `StatusMessageWatcherSync`.
+
+        Args:
+            run_client: The client for run that will be used to get a status and message.
+            to_logger: The logger to which the status message will be redirected.
+            check_period: The period with which the status message will be polled.
+        """
+        super().__init__(to_logger=to_logger, check_period=check_period)
+        self._run_client = run_client
+        self._logging_thread: Thread | None = None
+        self._stop_logging = False
+
+    def start(self) -> Thread:
+        """Start the logging thread. The caller has to handle any cleanup by manually calling the `stop` method."""
+        if self._logging_thread:
+            raise RuntimeError('Logging thread already active')
+        self._stop_logging = False
+        self._logging_thread = threading.Thread(target=self._log_changed_status_message)
+        self._logging_thread.start()
+        return self._logging_thread
+
+    def stop(self) -> None:
+        """Signal the _logging_thread thread to stop logging and wait for it to finish."""
+        if not self._logging_thread:
+            raise RuntimeError('Logging thread is not active')
+        time.sleep(self._final_sleep_time_s)
+        self._stop_logging = True
+        self._logging_thread.join()
+        self._logging_thread = None
+        self._stop_logging = False
+
+    def __enter__(self) -> Self:
+        """Start the logging task within the context. Exiting the context will cancel the logging task."""
+        self.start()
+        return self
+
+    def __exit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> None:
+        """Cancel the logging task."""
+        self.stop()
+
+    def _log_changed_status_message(self) -> None:
+        while True:
+            if not self._log_run_data(self._run_client.get()):
+                break
+            if self._stop_logging:
+                break
+            time.sleep(self._check_period)