#12 - add comments to properties to complete the documentation of AMQ Adapter

Reviewed-on: #38

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
+        _time_window_millis = self.config.backpressure.time_window
-        _idle_duration_millis = 1000 * self.config.backpressure.idle_duration
+        _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",
+                    "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s",
-                    self.current_parallel_executions,
+                    _current_cpu_usage,
-                )
+                    _average_cpu_usage,
-                logging_info(
+                    _cpu_usage_state,
-                    "Backpressure: Last data message time=%s, eventTime=%s",
                     self.last_data_message_time,
+                    self.last_backpressure_event,
                     self.last_backpressure_event_time,
                 )
-                logging_info(
+                if _average_cpu_usage < IDLE_THRESHOLD:
-                    "Backpressure: Last backpressure event=%s",
+                    if _cpu_usage_state != CPU_IDLE:
-                    self.last_backpressure_event,
+                        _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
+                    _cpu_usage_state == CPU_OVERLOAD
-                    and time.time() - self.last_backpressure_event_time
+                    and _now - _cpu_usage_changed > _overload_duration_millis
-                    > _time_window_sec
+                    and (
-                    and self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
+                        _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
+                    _cpu_usage_state == CPU_IDLE
-                    and time.time() - self.last_data_message_time
+                    and _now - _cpu_usage_changed > _idle_duration_millis
-                    >= _idle_duration_millis
+                    and _now - self.last_data_message_time > _idle_duration_millis
-                    and self.last_backpressure_event != ScalingRequestAlert.IDLE
+                    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 (
+                    if _now - self.last_backpressure_event_time > _time_window_millis:
-                        self.current_parallel_executions > 0
-                        and time.time() - self.last_backpressure_event_time
-                        > _time_window_sec
-                    ):
                         # Trigger the update event
                         _scaling_request: ScaleRequestV1 = ScaleRequestV1(
                             self.swarm_service_id,
                             self.swarm_task_id,
-                            self.parallel_workers,
+                            100,
-                            self.parallel_workers - self.current_parallel_executions,
+                            _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())
 

amqp/config/application.properties

@@ -2,30 +2,41 @@
 # CleverMicro AMQ Adapter settings
 # ====================================================================
 [CleverMicro-AMQ]
-cm.amq-adapter.service-name=amq-adapter-python
+# CleverMicro AMQ Adapter settings that identify this AMQ Adapter instance to other adapters
-cm.amq-adapter.generator-id=164
+
-cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:cleverswarm.localhost.#:5:0
+# A name to identify this service for the AMQ Adapter
+cm.amq-adapter.service-name=cleverthis-service-name
+# The CleverMicro AMQ Adapter instance ID, used to identify this instance in the CleverMicro AMQ Adapter cluster
+cm.amq-adapter.generator-id=1
+# Route mapping. For route detail description see section 3.1 in Onboarding Guide for Python:
+# https://docs.cleverthis.com/en/architecture/microservices/onboarding/python
+# NOTE: this is a comma-separated list of route mappings (multiple, comma-separated routes are allowed here)
+cm.amq-adapter.route-mapping=
+# Indicate if AMQ Adapter should respond with HTTP 401 Unauthorized if the request does not contain valid JWT token
 cm.amq-adapter.require-authenticated-user=false
 
+# CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding
 cm.dispatch.use-dlq=true
-cm.dispatch.use-confirms=false
 cm.dispatch.amq-host=rabbitmq
 cm.dispatch.amq-port=5672
 cm.dispatch.amq-port-tls=5671
 cm.dispatch.download-dir=/tmp/downloaded_files
-# Not used for CleverSwarm nor CleverBRAG or in tight coupling mode.
+# Not used for CleveSwarm nor CleverBRAG or in tight coupling mode.
 # applicable only in loose coupling mode, specify the destination where to forward the REST request
-#cm.dispatch.service-host=cleverswarm.localhost
+#cm.dispatch.service-host=cleverthis-service-name.localhost
 #cm.dispatch.service-port=8080
-#
+
+# Backpressure monitor settings
 # Define maximum number of messages of being processed in parallel before triggering backpressure OVERLOAD alert
 cm.backpressure.threshold=5
-# Define maximum CPU usage before triggering backpressure OVERLOAD alert
+# Define maximum CPU usage (%) before triggering backpressure OVERLOAD alert
-cm.backpressure.threshold-cpu=90
+cm.backpressure.threshold-cpu-overload=90
+# Define maximum CPU usage (%) before triggering backpressure IDLE alert
+cm.backpressure.threshold-cpu-idle=10
 # Define the time window between the Backpressure reports, in seconds
 cm.backpressure.time-window=10
 # Define the time window length, in seconds, during which, if CPU usage is consistently above the threshold, the system will trigger OVERLOAD alert
 cm.backpressure.cpu-overload-duration=30
 # Define the time window length, in seconds, during which, if no request is made to the Service, the system will trigger IDLE alert
-cm.backpressure.idle-duration=30
+cm.backpressure.cpu-idle-duration=30
 

amqp/service/amq_message_handler.py

@@ -75,12 +75,21 @@ class DataMessageHandler:
         self.file_handler: FileHandler = file_handler
         self.backpressure_handler: BackpressureHandler = backpressure_handler
 
-    async def inbound_data_message_callback(self, message: AbstractIncomingMessage):
+    async def inbound_data_message_callback_as_thread(
+        self, message: AbstractIncomingMessage
+    ):
+        """
+        The callback method that is called when a DataMessage is received from RabbitMQ, starts new Thread.
+        """
         threading.Thread(
-            target=lambda: asyncio.run(self.run_callback_thread(message))
+            target=lambda: asyncio.run(
+                self.inbound_data_message_callback_no_thread(message)
+            )
         ).start()
 
-    async def run_callback_thread(self, message: AbstractIncomingMessage):
+    async def inbound_data_message_callback_no_thread(
+        self, message: AbstractIncomingMessage
+    ):
         """
         Receives the DataMessage as bytes from RabbitMQ queue and deserializes it to the DataMessage object
         Ensures Jaeger trace-id propagation.

amqp/service/amq_service.py

@@ -107,7 +107,7 @@ class AMQService:
             try:
                 await self.rabbit_mq_client.register_inbound_routes(
                     route_mapping,
-                    self.amq_data_message_handler.inbound_data_message_callback,
+                    self.amq_data_message_handler.inbound_data_message_callback_no_thread,
                 )
                 return await self.rabbit_mq_client.register_data_reply_callback(
                     self.amq_data_message_handler.reply_received_callback

cleverswarm/clever_swarm_adapter.py

@@ -252,11 +252,13 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
                 for _body_param in _route.dependant.body_params:
                     if _body_param.name in _form_data:
                         _file_data = _form_data[_body_param.name]
-                        actual_filepath = message.extra_data[_body_param.name]
                         _form_data[_body_param.name] = (
                             _body_param.type_.__name__ == "UploadFile"
+                            and message.extra_data.get(_body_param.name)
                             and UploadFile(
-                                file=pathlib.Path(actual_filepath).open("rb"),
+                                file=pathlib.Path(
+                                    message.extra_data.get(_body_param.name)
+                                ).open("rb"),
                                 size=_file_data["size"],
                                 filename=_file_data["filename"],
                                 headers=Headers(

pyproject.toml

@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.23-dev"
+version = "0.2.27"
-description = "The Python implementation of the AMQ Adaptor for CleverMicro"
+description = "The Python implementation of the AMQ Adapter for CleverMicro"
 readme = "README.md"
 authors = [
     { name = "Stanislav Hejny", email = "stanislav.hejny@cleverthis.com" }
@@ -21,13 +21,14 @@ requires-python = ">=3.11"
 dependencies = [
     "aio-pika~=9.5.5",
     "aiohttp~=3.11.11",
-    "aio_pika~=9.5.5",
     "fastapi==0.114.2",
     "opentelemetry-api",
     "opentelemetry-sdk",
     "opentelemetry-exporter-otlp",
     "opentelemetry-exporter-prometheus",
     "opentelemetry-instrumentation-pika",
+    "pika~=1.3.2",
+    "psutil",
     "python_multipart"
 ]
 

tools/create_exchange.py

@@ -0,0 +1,90 @@
+import asyncio
+import aio_pika
+import os
+import argparse
+import sys
+
+from aio_pika.abc import AbstractRobustExchange
+
+
+async def create_exchange(loop, rabbit_user, rabbit_password, rabbit_host):
+    """
+    Asynchronously connects to RabbitMQ, declares a direct exchange,
+    and prints the result.  Handles potential connection errors.
+    """
+    exchange_name = "cleverthis.clevermicro.management"
+    connection = None  # Keep track of the connection
+
+    try:
+        # Construct the connection URI from environment variables
+        connection_uri = f"amqp://{rabbit_user}:{rabbit_password}@{rabbit_host}/"
+
+        # Attempt to establish the connection
+        connection = await aio_pika.connect_robust(connection_uri, loop=loop)
+        print(f"Connected to RabbitMQ: {connection_uri}")
+
+        # Create a channel
+        channel = await connection.channel()
+
+        # Declare the exchange
+        exchange: AbstractRobustExchange = await channel.declare_exchange(
+            exchange_name, aio_pika.ExchangeType.DIRECT
+        )
+        print(f"Exchange '{exchange.name}' created successfully.")
+
+    except aio_pika.exceptions.AMQPConnectionError as e:
+        print(f"Error: Could not connect to RabbitMQ.  Details: {e}")
+        sys.exit(1)  # Exit with an error code
+    except aio_pika.exceptions.AMQPChannelError as e:
+        print(f"Error creating channel or exchange: {e}")
+        sys.exit(1)
+    except Exception as e:
+        print(f"An unexpected error occurred: {e}")
+        sys.exit(1)
+    finally:
+        # Ensure the connection is closed, even if errors occur
+        if connection:
+            await connection.close()
+            print("Connection to RabbitMQ closed.")
+
+def main():
+    """
+    Main function to parse command-line arguments, retrieve RabbitMQ
+    credentials from environment variables, and run the
+    async exchange creation.
+    """
+    # Set up argument parsing
+    parser = argparse.ArgumentParser(description="Create a RabbitMQ exchange.")
+    parser.add_argument(
+        "--host",
+        default="localhost",
+        help="RabbitMQ host (default: localhost)",
+    )
+
+    # Parse the arguments
+    args = parser.parse_args()
+    rabbit_host = args.host
+
+    # Get RabbitMQ credentials from environment variables
+    rabbit_user = os.environ.get("RABBIT_MQ_USER")
+    rabbit_password = os.environ.get("RABBIT_MQ_PASSWORD")
+
+    if not rabbit_user or not rabbit_password:
+        print(
+            "Error: RabbitMQ credentials not found in environment variables.\n"
+            "Please set RABBIT_MQ_USER and RABBIT_MQ_PASSWORD."
+        )
+        sys.exit(1)  # Exit with an error code
+
+    # Create the event loop
+    loop = asyncio.get_event_loop()
+
+    # Run the asynchronous exchange creation
+    loop.run_until_complete(create_exchange(loop, rabbit_user, rabbit_password, rabbit_host))
+
+    # Close the loop
+    loop.close()
+
+
+if __name__ == "__main__":
+    main()