feat \#16 - add unittest for backpressure_handler and logging_utils

amqp/adapter/backpressure_handler.py

@@ -48,8 +48,15 @@ class ScaleRequestV1:
 
 
 class RunningAverage:
-    def __init__(self, num_items):
-        self.buffer = [0] * num_items  # Fixed-size array
+    """
+    A class to maintain a running average of the last N values.
+    The intended use is to track CPU usage over time window, that is directly related to sample rate and window size.
+    This is the reason why the constructor takes the number of items to store, calculated as window size / sample rate
+    Initial_value is for unit test to emulate OVERLOAD or IDLE condition.
+    """
+
+    def __init__(self, num_items, initial_value=0):
+        self.buffer = [initial_value] * num_items  # Fixed-size array
         self.pointer = 0  # Current position in array
         self.num_items = num_items  # Maximum number of items to store
         self.is_filled = False  # Track if buffer is fully filled
@@ -89,6 +96,8 @@ class BackpressureHandler:
     # helps prevent flooding the system with backpressure events
     last_backpressure_event_time = 0
     last_backpressure_event = ScalingRequestAlert.UPDATE
+    # _callback_list is used in unit tests to record the invoked callbacks
+    _callback_list = None
 
     def __init__(
         self,
@@ -105,6 +114,18 @@ class BackpressureHandler:
         self.swarm_task_id = self.config.amq_adapter.swarm_task_id
         self.parallel_workers = self.config.backpressure.threshold_threads
         self.average_cpu_usage = None
+        self.do_loop = -1
+
+    def _do_loop(self) -> bool:
+        """
+        Check if the loop should continue running.
+        Positive value of do_loop indicates the number of iterations left.
+        Negative value indicates the loop should run indefinitely.
+        """
+        _val = self.do_loop != 0
+        if self.do_loop > 0:
+            self.do_loop -= 1
+        return _val
 
     def increase_parallel_executions(self):
         """Increase the number of parallel executions"""
@@ -154,93 +175,96 @@ class BackpressureHandler:
                 self.last_backpressure_event_time = time.time()
                 self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
 
-    def backpressure_monitor_loop(self):
+    async def _backpressure_monitor(self):
+        """Monitor the backpressure conditions and trigger events accordingly"""
         _time_window_millis = self.config.backpressure.time_window
         _idle_duration_millis = self.config.backpressure.idle_duration
-        _overload_duration_millis = self.config.backpressure.idle_duration
-        _loop = asyncio.new_event_loop()
-        _monitor_interval = 0.2  # Monitor every 200ms
+        _overload_duration_millis = self.config.backpressure.cpu_overload_duration
+        _monitor_interval = 0.5  # Monitor every 500ms
         CPU_IDLE, CPU_ACTIVE, CPU_OVERLOAD = 0, 1, 2
         IDLE_THRESHOLD, OVERLOAD_THRESHOLD = 15, 90
-        self.average_cpu_usage = RunningAverage(
-            int(max(_overload_duration_millis, _idle_duration_millis) // _monitor_interval)
-        )
+        if self.average_cpu_usage is None:
+            self.average_cpu_usage = RunningAverage(
+                int(
+                    max(_overload_duration_millis, _idle_duration_millis)
+                    // _monitor_interval
+                )
+            )
+        _cpu_usage_state = CPU_ACTIVE
+        _cpu_usage_changed = time.time()
+        while self._do_loop():
+            _current_cpu_usage = psutil.cpu_percent(interval=None)
+            self.average_cpu_usage.add(_current_cpu_usage)
+            _average_cpu_usage = self.average_cpu_usage.get_average()
+            _now = time.time()
+            # logging_info(
+            #    "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s",
+            #    _current_cpu_usage,
+            #    _average_cpu_usage,
+            #    _cpu_usage_state,
+            #    self.last_data_message_time,
+            #    self.last_backpressure_event,
+            #    self.last_backpressure_event_time,
+            # )
+            if _average_cpu_usage < IDLE_THRESHOLD:
+                if _cpu_usage_state != CPU_IDLE:
+                    _cpu_usage_changed = _now
+                _cpu_usage_state = CPU_IDLE
+            elif _average_cpu_usage > OVERLOAD_THRESHOLD:
+                if _cpu_usage_state != CPU_OVERLOAD:
+                    _cpu_usage_changed = _now
+                _cpu_usage_state = CPU_OVERLOAD
+            else:
+                if _cpu_usage_state != CPU_ACTIVE:
+                    _cpu_usage_changed = _now
+                _cpu_usage_state = CPU_ACTIVE
 
-        async def _backpressure_monitor():
-            """Monitor the backpressure conditions and trigger events accordingly"""
-            _cpu_usage_state = CPU_ACTIVE
-            _cpu_usage_changed = time.time()
-            while True:
-                _current_cpu_usage = psutil.cpu_percent(interval=None)
-                self.average_cpu_usage.add(_current_cpu_usage)
-                _average_cpu_usage = self.average_cpu_usage.get_average()
-                _now = time.time()
-                # logging_info(
-                #    "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s",
-                #    _current_cpu_usage,
-                #    _average_cpu_usage,
-                #    _cpu_usage_state,
-                #    self.last_data_message_time,
-                #    self.last_backpressure_event,
-                #    self.last_backpressure_event_time,
-                # )
-                if _average_cpu_usage < IDLE_THRESHOLD:
-                    if _cpu_usage_state != CPU_IDLE:
-                        _cpu_usage_changed = _now
-                    _cpu_usage_state = CPU_IDLE
-                elif _average_cpu_usage > OVERLOAD_THRESHOLD:
-                    if _cpu_usage_state != CPU_OVERLOAD:
-                        _cpu_usage_changed = _now
-                    _cpu_usage_state = CPU_OVERLOAD
-                else:
-                    if _cpu_usage_state != CPU_ACTIVE:
-                        _cpu_usage_changed = _now
-                    _cpu_usage_state = CPU_ACTIVE
-
-                # Check if the current time exceeds the last backpressure event time
-                if (
-                    _cpu_usage_state == CPU_OVERLOAD
-                    and _now - _cpu_usage_changed > _overload_duration_millis
-                    and (
-                        _now - self.last_backpressure_event_time > _time_window_millis
-                        or self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
+            # Check if the current time exceeds the last backpressure event time
+            if (
+                _cpu_usage_state == CPU_OVERLOAD
+                and _now - _cpu_usage_changed > _overload_duration_millis
+                and (
+                    _now - self.last_backpressure_event_time > _time_window_millis
+                    or self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
+                )
+            ):
+                # Trigger the overload event
+                await self.handle_backpressure_overload_event()
+                self.last_backpressure_event_time = _now
+                self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
+            elif (
+                _cpu_usage_state == CPU_IDLE
+                and _now - _cpu_usage_changed > _idle_duration_millis
+                and _now - self.last_data_message_time > _idle_duration_millis
+                and (
+                    _now - self.last_backpressure_event_time > _time_window_millis
+                    or self.last_backpressure_event != ScalingRequestAlert.IDLE
+                )
+            ):
+                # Trigger the idle event
+                await self._handle_backpressure_idle_event()
+                self.last_backpressure_event = ScalingRequestAlert.IDLE
+                self.last_backpressure_event_time = _now
+            else:
+                # Trigger update event
+                if _now - self.last_backpressure_event_time > _time_window_millis:
+                    # Trigger the update event
+                    _scaling_request: ScaleRequestV1 = ScaleRequestV1(
+                        self.swarm_service_id,
+                        self.swarm_task_id,
+                        100,
+                        _average_cpu_usage,
+                        ScalingRequestAlert.UPDATE,
                     )
-                ):
-                    # Trigger the overload event
-                    await self.handle_backpressure_overload_event()
+                    # Address the message to any adapter capable of supporting BACKPRESSURE request
+                    await self.publish_backpressure_request(_scaling_request)
                     self.last_backpressure_event_time = _now
-                    self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
-                elif (
-                    _cpu_usage_state == CPU_IDLE
-                    and _now - _cpu_usage_changed > _idle_duration_millis
-                    and _now - self.last_data_message_time > _idle_duration_millis
-                    and (
-                        _now - self.last_backpressure_event_time > _time_window_millis
-                        or self.last_backpressure_event != ScalingRequestAlert.IDLE
-                    )
-                ):
-                    # Trigger the idle event
-                    await self._handle_backpressure_idle_event()
-                    self.last_backpressure_event = ScalingRequestAlert.IDLE
-                    self.last_backpressure_event_time = _now
-                else:
-                    # Trigger update event
-                    if _now - self.last_backpressure_event_time > _time_window_millis:
-                        # Trigger the update event
-                        _scaling_request: ScaleRequestV1 = ScaleRequestV1(
-                            self.swarm_service_id,
-                            self.swarm_task_id,
-                            100,
-                            _average_cpu_usage,
-                            ScalingRequestAlert.UPDATE,
-                        )
-                        # Address the message to any adapter capable of supporting BACKPRESSURE request
-                        await self.publish_backpressure_request(_scaling_request)
-                        self.last_backpressure_event_time = _now
-                        self.last_backpressure_event = ScalingRequestAlert.UPDATE
-                await asyncio.sleep(_monitor_interval)
+                    self.last_backpressure_event = ScalingRequestAlert.UPDATE
+            await asyncio.sleep(_monitor_interval)
 
-        _loop.run_until_complete(_backpressure_monitor())
+    def backpressure_monitor_loop(self):
+        _loop = asyncio.new_event_loop()
+        _loop.run_until_complete(self._backpressure_monitor())
 
     async def handle_backpressure_overload_event(self):
         logging_warning("Backpressure: Capacity close to depleted!")
@@ -283,6 +307,8 @@ class BackpressureHandler:
                 _exchange = await channel.get_exchange(name="cleverthis.clevermicro.management")
                 return _exchange
 
+            if BackpressureHandler._callback_list is not None:
+                BackpressureHandler._callback_list["_wrap_rabbit_mq_api_init"] = 1
             self.exchange = await await_result(
                 asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api_init(self.channel), self.loop)
             )
@@ -312,4 +338,8 @@ class BackpressureHandler:
                     return True
                 return False
 
-            await await_future(asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop))
+            if BackpressureHandler._callback_list is not None:
+                BackpressureHandler._callback_list["_wrap_rabbit_mq_api"] = 1
+            await await_future(
+                asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop)
+            )

amqp/adapter/logging_utils.py

@@ -120,7 +120,7 @@ def logging_debug(msg: str, *args) -> None:
     _thread_id = threading.get_ident()
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
-        logging.info(
+        logging.debug(
             f"{_thread_id}:{__version__} [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )