Merge pull request #11 from gnir-work/feature/add-loop-lag-metric

Feature/add loop lag metric

Author: gnir-work

.github/workflows/test_and_lint_package.yaml

@@ -14,7 +14,7 @@ jobs:
       fail-fast: true
       matrix:
         os: [ "ubuntu-latest" ]
-        python-version: [ "3.8", "3.9", "3.10", "3.11", "3.12" ]
+        python-version: [ "3.9", "3.10", "3.11", "3.12" ]
     runs-on: ${{ matrix.os }}
     steps:
       - name: Check out repository

.gitignore

@@ -158,4 +158,6 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
-junit/
+junit/
+
+.DS_Store

Makefile

@@ -20,4 +20,7 @@ lint: lint-formatters lint-static-code-analysis
 pre-commit: format lint-static-code-analysis
 
 test:
-	poetry run pytest --cov=monitored_ioloop -n auto tests
+	poetry run pytest --cov=monitored_ioloop -n auto tests
+
+test-for-flakiness:
+	poetry run pytest --cov=monitored_ioloop -n auto --count=10

README.md

@@ -48,9 +48,9 @@ The usage is the same as the asyncio event loop, but with `monitored_ioloop.moni
 ## The monitor callback
 The monitor callback will be called for every execution that the event loop initiates.  
 With every call you will receive an [IoLoopMonitorState](monitored_ioloop/monitoring.py) object that contains the following information:
-- `wall_loop_duration`: The time it took to execute the current step of the event loop in wall time.
-- `cpu_loop_duration`: The time it took to execute the current step of the event loop in CPU time.
-- `handles_count`: The amount of handles (think about them as tasks) that the IO loop is currently handling.
+- `callback_wall_time`: Wall executing time of the callback.
+- `loop_handles_count`: The amount of handles (think about them as tasks) that the IO loop is currently handling.
+- `loop_lag`: The amount of time it took from the moment the task was added to the loop until it was executed.
 
 ## Performance impact
 As many of you might be concerned about the performance impact of this library, I have run some benchmarks to measure the performance impact of this library.  
@@ -65,11 +65,14 @@ Currently there is only the [fastapi with prometheus exporter example](examples/
 ## Roadmap
 - [x] Add support for the amount of `Handle`'s on the event loop
 - [x] Add an examples folder
+- [x] Add loop lag metric (Inspired from nodejs loop monitoring)
 - [ ] Add visibility into which `Handle` are making the event loop slower
-- [ ] Add easier integration with popular monitoring tools like Prometheus
 - [ ] Add easier integration with `uvicorn`
+- [ ] Add easier integration with popular monitoring tools like Prometheus
+
 
 
 ## Credits
-I took a lot of inspiration from the [uvloop](https://github.com/MagicStack/uvloop) project with everythin
-regarding the user interface of swapping the IO loop.
+* I took a lot of inspiration from the [uvloop](https://github.com/MagicStack/uvloop) project with everything
+regarding the user interface of swapping the IO loop.
+* The great pycon talk - https://www.youtube.com/watch?v=GSiZkP7cI80&t=16s

examples/fastapi_with_prometheus/server.py

@@ -34,7 +34,7 @@ async def blocking_slow(sleep_for: int) -> str:
 
 
 def monitor_ioloop(ioloop_monitor_state: IoLoopMonitorState) -> None:
-    ioloop_execution_time_histogram.observe(ioloop_monitor_state.wall_loop_duration)
+    ioloop_execution_time_histogram.observe(ioloop_monitor_state.callback_wall_time)
 
 
 def main() -> None:

examples/simple_python_example/simple_python_example.py

@@ -0,0 +1,53 @@
+from monitored_ioloop.monitored_asyncio import MonitoredAsyncIOEventLoopPolicy
+from monitored_ioloop.monitoring import IoLoopMonitorState
+import asyncio
+import logging
+import time
+
+logger = logging.getLogger("simple_python_example")
+
+
+async def main() -> None:
+    async def sleep(coroutine_id: int) -> None:
+        logger.info(f"[id: {coroutine_id}] Before non-blocking sleep")
+        await asyncio.sleep(1)
+        logger.info(f"[id: {coroutine_id}] After non-blocking sleep")
+
+    async def blocking_sleep(coroutine_id: int) -> None:
+        logger.info(f"[id: {coroutine_id}] Before blocking sleep")
+        time.sleep(1)
+        logger.info(f"[id: {coroutine_id}] After blocking sleep")
+
+    await asyncio.gather(
+        sleep(coroutine_id=0),
+        blocking_sleep(coroutine_id=1),
+        sleep(coroutine_id=2),
+        sleep(coroutine_id=3),
+        blocking_sleep(coroutine_id=4),
+        sleep(coroutine_id=5),
+    )
+
+
+def monitor(ioloop_monitor_state: IoLoopMonitorState) -> None:
+    if ioloop_monitor_state.callback_wall_time > 0.1:
+        logger.warning(
+            f"Blocking operation detected, executing took: {ioloop_monitor_state.callback_wall_time}"
+        )
+    if ioloop_monitor_state.loop_lag > 0.1:
+        logger.warning(
+            f"A task was executed after a significant delay: {ioloop_monitor_state.loop_lag}"
+        )
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        datefmt="[%d-%b-%y %H:%M:%S]",
+        format="%(asctime)s %(message)s",
+    )
+    logger.info("starting")
+    asyncio.set_event_loop_policy(
+        MonitoredAsyncIOEventLoopPolicy(monitor_callback=monitor)
+    )
+    logger.info("Set event loop")
+    asyncio.run(main())

monitored_ioloop/monioted_ioloop_base.py

@@ -16,8 +16,6 @@ def __init__(
         # as abstract in typeshed, we have to put them in so mypy thinks
         # the base methods are overridden. This is the same approach taken
         # for the Windows event loop policy classes in typeshed.
-        def get_child_watcher(self) -> typing.NoReturn:
-            ...
+        def get_child_watcher(self) -> typing.NoReturn: ...
 
-        def set_child_watcher(self, watcher: typing.Any) -> typing.NoReturn:
-            ...
+        def set_child_watcher(self, watcher: typing.Any) -> typing.NoReturn: ...

monitored_ioloop/monitored_asyncio.py

@@ -24,7 +24,7 @@ def __init__(
     def call_soon(
         self,
         callback: typing.Callable[..., typing.Any],
-        *args: typing.Any,
+        *args: typing.Any,  # type: ignore
         **kwargs: typing.Any,
     ) -> Handle:
         callback_with_monitoring = wrap_callback_with_monitoring(

monitored_ioloop/monitored_uvloop.py

@@ -31,7 +31,7 @@ def __init__(
     def call_soon(
         self,
         callback: typing.Callable[..., typing.Any],
-        *args: typing.Any,
+        *args: typing.Any,  # type: ignore
         **kwargs: typing.Any,
     ) -> Handle:
         callback_with_monitoring = wrap_callback_with_monitoring(

monitored_ioloop/monitoring.py

@@ -9,12 +9,34 @@
 
 @dataclass
 class IoLoopMonitorState:
-    # The time it took to execute the loop in wall time (For example asyncio.sleep will also be recorded).
-    wall_loop_duration: float
-    # The time it took to execute the loop in cpu time (For example asyncio.sleep will not be recorded).
-    cpu_loop_duration: float
-    # The amount of handles in the loop - https://docs.python.org/3/library/asyncio-eventloop.html#callback-handles
-    handles_count: int
+    """
+    A dataclass containing the state of the loop when the callback was executed.
+    This class is the interface that the monitor callback will receive.
+
+    A basic Lexicon:
+    * Handle - A wrapper for a callback that is scheduled to be executed by the loop.
+    * Callback - The function that is executed by the loop.
+    * Loop - The event loop that is executing the callbacks.
+    """
+
+    """
+    Wall executing time of the callback
+    It can be the whole coroutine or parts of it, depending on if the executing control
+    was delegated back the loop or not.
+
+    Wall Time explanation - https://en.wikipedia.org/wiki/Wall-clock_time
+    """
+    callback_wall_time: float
+
+    """
+    The amount of handles in the loop, excluding the current one.
+    """
+    loop_handles_count: int
+
+    """
+    The amount of time it took from the moment the coroutine was added to the loop until it was executed.
+    """
+    loop_lag: float
 
 
 @dataclass
@@ -48,21 +70,21 @@ def wrap_callback_with_monitoring(
     back to the monitor_callback.
     """
     ioloop_state.increase_handles_count_thread_safe(1)
+    added_to_loop_time = time.perf_counter()
 
     def wrapper(*inner_args: typing.Any, **inner_kwargs: typing.Any) -> typing.Any:
-        start_wall_time = time.time()
-        start_cpu_time = time.process_time()
+        loop_lag = time.perf_counter() - added_to_loop_time
+        start_wall_time = time.perf_counter()
         response = callback(*inner_args, **inner_kwargs)
         ioloop_state.decrease_handles_count_thread_safe(1)
-        wall_duration = time.time() - start_wall_time
-        cpu_duration = time.process_time() - start_cpu_time
+        wall_duration = time.perf_counter() - start_wall_time
 
         try:
             monitor_callback(
                 IoLoopMonitorState(
-                    wall_loop_duration=wall_duration,
-                    cpu_loop_duration=cpu_duration,
-                    handles_count=ioloop_state.handles_count,
+                    callback_wall_time=wall_duration,
+                    loop_handles_count=ioloop_state.handles_count,
+                    loop_lag=loop_lag,
                 )
             )
         except Exception: