create heartbeat component

Author: GabrielSalla

docs/monitoring_sentinela.md

@@ -48,5 +48,6 @@ The Prometheus metrics provided by Sentinela are:
     - Labels: `action_name`
 - `executor_request_execution_seconds`: Summary - Time to run the request
     - Labels: `action_name`
+- `heartbeat_average_time`: Gauge - Average time between heartbeats in seconds
 - `registry_monitors_ready_timeout_count`: Counter - Count of times the application timed out waiting for monitors to be ready
 - `registry_monitor_not_registered_count`: Counter - Count of times a monitor is not registered after a load attempt

src/components/heartbeat/__init__.py

@@ -0,0 +1,5 @@
+from .heartbeat import run
+
+__all__ = [
+    "run",
+]

src/components/heartbeat/heartbeat.py

@@ -0,0 +1,49 @@
+import logging
+import time
+from collections import deque
+from itertools import pairwise
+
+import prometheus_client
+
+import utils.app as app
+from configs import configs
+
+_logger = logging.getLogger("heartbeat")
+
+prometheus_heartbeat_average_time = prometheus_client.Gauge(
+    "heartbeat_average_time", "Average time between heartbeats in seconds"
+)
+
+
+def _is_heartbeat_delayed(timestamps: deque[float], threshold: float) -> bool:
+    """Determine if the heartbeat is delayed based on the average latency between timestamps"""
+    if len(timestamps) < 2:
+        return False
+
+    latencies = [b - a for a, b in pairwise(timestamps)]
+    average_latency = sum(latencies) / len(latencies)
+    prometheus_heartbeat_average_time.set(average_latency)
+    return average_latency > threshold
+
+
+async def run() -> None:
+    """Create a heartbeat for the application to detect when some tasks are not yielding control
+    back to the event loop. If the heartbeat is delayed, a warning message is logged."""
+    timestamps = deque[float](maxlen=10)
+    last_warning_timestamp = 0.0
+
+    while app.running():
+        timestamp = time.time()
+        timestamps.append(timestamp)
+        heartbeat_delayed = _is_heartbeat_delayed(timestamps, configs.heartbeat_time * 1.05)
+
+        # Prevent warning messages from being sent too frequently
+        can_warn = timestamp - last_warning_timestamp > 10
+        if can_warn and heartbeat_delayed:
+            _logger.warning(
+                "High average heartbeat interval. "
+                "Blocking operations are preventing tasks from executing"
+            )
+            last_warning_timestamp = timestamp
+
+        await app.sleep(configs.heartbeat_time)

tests/components/heartbeat/test_heartbeat.py

@@ -0,0 +1,63 @@
+import asyncio
+import time
+from collections import deque
+from unittest.mock import MagicMock
+
+import pytest
+
+import components.heartbeat.heartbeat as heartbeat
+import utils.app as app
+from configs import configs
+
+pytestmark = pytest.mark.asyncio(loop_scope="session")
+
+
+@pytest.mark.parametrize(
+    "timestamps, threshold, expected_result",
+    [
+        (deque([100]), 1, False),
+        (deque([100, 101]), 1, False),
+        (deque([100, 102]), 1, True),
+        (deque([1, 3, 5, 7, 9]), 2, False),
+        (deque([1, 3, 5, 7, 10]), 2, True),
+    ],
+)
+async def test_is_heartbeat_delayed(timestamps, threshold, expected_result):
+    """'is_heartbeat_delayed' should return True when average latency exceeds threshold"""
+    assert heartbeat._is_heartbeat_delayed(timestamps, threshold) is expected_result
+
+
+async def test_run(mocker, monkeypatch):
+    """'run' should append the current timestamp while app is running"""
+    monkeypatch.setattr(configs, "heartbeat_time", 0.05)
+    heartbeat_logger_warning_spy: MagicMock = mocker.spy(heartbeat._logger, "warning")
+
+    task = asyncio.create_task(heartbeat.run())
+
+    await asyncio.sleep(0.1)
+    assert heartbeat_logger_warning_spy.call_count == 0
+    time.sleep(0.1)
+    await asyncio.sleep(0)
+    assert heartbeat_logger_warning_spy.call_count == 1
+
+    app.stop()
+    await asyncio.wait_for(task, timeout=0.1)
+
+
+async def test_run_cooldown(mocker, monkeypatch):
+    """'run' should respect the cooldown period between warnings when heartbeat is delayed"""
+    monkeypatch.setattr(configs, "heartbeat_time", 0.05)
+    heartbeat_logger_warning_spy: MagicMock = mocker.spy(heartbeat._logger, "warning")
+
+    task = asyncio.create_task(heartbeat.run())
+
+    await asyncio.sleep(0.1)
+    time.sleep(0.1)
+    await asyncio.sleep(0)
+    assert heartbeat_logger_warning_spy.call_count == 1
+    time.sleep(0.1)
+    await asyncio.sleep(0)
+    assert heartbeat_logger_warning_spy.call_count == 1
+
+    app.stop()
+    await asyncio.wait_for(task, timeout=0.1)