amqp/adapter/backpressure_handler.py

@@ -4,6 +4,7 @@ import time
 from asyncio import AbstractEventLoop
 from threading import Thread
 
+import psutil
 from aio_pika import Message
 from aio_pika.abc import AbstractRobustChannel
 
@@ -46,6 +47,40 @@ class ScaleRequestV1:
         )
 
 
+class RunningAverage:
+    def __init__(self, num_items):
+        self.buffer = [0] * 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
+
+    def add(self, value):
+        # Store the value at current pointer position
+        self.buffer[self.pointer] = value
+
+        # Move pointer to next position with wrap-around
+        self.pointer += 1
+        if self.pointer >= self.num_items:
+            self.pointer = 0
+            self.is_filled = True
+
+    def get_average(self):
+        # Determine how many items we should average
+        count = self.num_items if self.is_filled else self.pointer
+
+        if count == 0:
+            return 0.0  # Avoid division by zero
+
+        return sum(self.buffer) / count
+
+    def get_current_values(self):
+        """Returns the values in chronological order (oldest first)"""
+        if not self.is_filled:
+            return self.buffer[: self.pointer]
+
+        return self.buffer[self.pointer :] + self.buffer[: self.pointer]
+
+
 class BackpressureHandler:
     # Track the number of messages currently being processed
     current_parallel_executions = 0
@@ -69,6 +104,7 @@ class BackpressureHandler:
         self.swarm_service_id = self.config.amq_adapter.swarm_service_id
         self.swarm_task_id = self.config.amq_adapter.swarm_task_id
         self.parallel_workers = self.config.backpressure.threshold_threads
+        self.average_cpu_usage = None
 
     def increase_parallel_executions(self):
         """Increase the number of parallel executions"""
@@ -119,67 +155,93 @@ class BackpressureHandler:
                 self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
 
     def backpressure_monitor_loop(self):
-        _time_window_sec = self.config.backpressure.time_window
-        _idle_duration_millis = 1000 * self.config.backpressure.idle_duration
+        _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
+        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
+            )
+        )
 
         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: Monitor loop, current=%s",
-                    self.current_parallel_executions,
-                )
-                logging_info(
-                    "Backpressure: Last data message time=%s, eventTime=%s",
+                    "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,
                 )
-                logging_info(
-                    "Backpressure: Last backpressure event=%s",
-                    self.last_backpressure_event,
-                )
+                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 (
-                    self.current_parallel_executions >= self.parallel_workers
-                    and time.time() - self.last_backpressure_event_time
-                    > _time_window_sec
-                    and self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
+                    _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 = time.time()
+                    self.last_backpressure_event_time = _now
                     self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
                 elif (
-                    self.current_parallel_executions == 0
-                    and time.time() - self.last_data_message_time
-                    >= _idle_duration_millis
-                    and self.last_backpressure_event != ScalingRequestAlert.IDLE
+                    _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 = time.time()
+                    self.last_backpressure_event_time = _now
                 else:
                     # Trigger update event
-                    if (
-                        self.current_parallel_executions > 0
-                        and time.time() - self.last_backpressure_event_time
-                        > _time_window_sec
-                    ):
+                    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,
-                            self.parallel_workers,
-                            self.parallel_workers - self.current_parallel_executions,
+                            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 = time.time()
+                        self.last_backpressure_event_time = _now
                         self.last_backpressure_event = ScalingRequestAlert.UPDATE
-                await asyncio.sleep(_time_window_sec)
+                await asyncio.sleep(_monitor_interval)
 
         _loop.run_until_complete(_backpressure_monitor())
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.26"
+version = "0.2.27"
 description = "The Python implementation of the AMQ Adapter for CleverMicro"
 readme = "README.md"
 authors = [
@@ -28,6 +28,7 @@ dependencies = [
     "opentelemetry-exporter-prometheus",
     "opentelemetry-instrumentation-pika",
     "pika~=1.3.2",
+    "psutil",
     "python_multipart"
 ]