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

Reviewed-on: #38

.forgejo/workflows/publish-python.yaml

@@ -4,7 +4,6 @@ on:
     branches:
       - master # publish release on master
       - develop # publish snapshot on develop
-      - feat/* # test
   workflow_dispatch:
   # allow manual trigger
 env:
@@ -38,19 +37,10 @@ jobs:
       - name: Build wheel
         run: |
           python -m build
-      # upload to forgejo artifacts
-      # TODO: remove this when pulp registry is working
-      - name: upload artifacts
-        uses: https://code.forgejo.org/forgejo/upload-artifact@v4
-        with:
-          name: amq-adapter-dist.zip
-          path: dist/*
-          if-no-files-found: error
-          retention-days: 7
       # publish
       - name: publish to pulp pypi
         run: |
-          twine upload --repository-url https://pkg.cleverthis.io/pypi/clevermicro/amq-adapter-python/ \
+          twine upload --repository-url https://pkg.cleverthis.io/pypi/clevermicro/amq-adapter-python/simple/ \
                        --username $CI_REGISTRY_USER \
                        --password $CI_REGISTRY_PASSWORD \
                        --verbose \

amqp/adapter/backpressure_handler.py

@@ -1,16 +1,17 @@
 import asyncio
 import json
-import logging
-import os
-import threading
 import time
 from asyncio import AbstractEventLoop
+from threading import Thread
 
+import psutil
 from aio_pika import Message
-from aio_pika.abc import AbstractRobustChannel, ExchangeType
+from aio_pika.abc import AbstractRobustChannel
 
 from amqp.adapter.logging_utils import logging_info, logging_warning
+from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import ScalingRequestAlert
+from amqp.router.utils import await_future, await_result
 
 
 class ScaleRequestV1:
@@ -29,6 +30,7 @@ class ScaleRequestV1:
         self.maxAvailability = maxAvailability
         self.currentAvailability = currentAvailability
         self.requestType = requestType
+        self.thread = None
 
     def to_json(self) -> str:
         """Converts the object to a JSON string matching the Java structure"""
@@ -45,29 +47,203 @@ 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
-    last_data_message_time = 0  # helps detect IDLE condition
-    last_backpressure_event_time = (
-        0  # helps prevent flooding the system with backpressure events
-    )
-    last_backpressure_event = ScalingRequestAlert.IDLE
+    # helps detect IDLE condition
+    last_data_message_time = 0
+    # helps prevent flooding the system with backpressure events
+    last_backpressure_event_time = 0
+    last_backpressure_event = ScalingRequestAlert.UPDATE
 
-    parallel_workers = int(os.getenv("PARALLEL_WORKERS", default=0))
-    swarm_service_id = os.getenv("SWARM_SERVICE_ID", default="clever-amqp-service")
-    swarm_task_id = os.getenv("SWARM_TASK_ID", default="python-adapter")
-
-    def __init__(self, channel: AbstractRobustChannel, loop: AbstractEventLoop):
+    def __init__(
+        self,
+        channel: AbstractRobustChannel,
+        loop: AbstractEventLoop,
+        config: AMQConfiguration,
+    ):
+        self.lock = None
         self.channel = channel
         self.loop = loop
+        self.config = config
         self.exchange = None
+        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 handle_backpressure(self, scaling_request: ScaleRequestV1):
-        # Handle the backpressure logic here
-        print(
-            f"Handling backpressure for {scaling_request.serviceId} with request type {scaling_request.requestType}"
+    def increase_parallel_executions(self):
+        """Increase the number of parallel executions"""
+        logging_info(
+            "Backpressure: Increase parallel executions, current=%s",
+            self.current_parallel_executions,
         )
+        self.current_parallel_executions += 1
+
+    def decrease_parallel_executions(self):
+        """Decrease the number of parallel executions"""
+        logging_info(
+            "Backpressure: Decrease parallel executions, current=%s",
+            self.current_parallel_executions,
+        )
+        if self.current_parallel_executions > 0:
+            self.current_parallel_executions -= 1
+
+    def update_parallel_executions(self, count: int):
+        """Update the number of parallel executions"""
+        self.current_parallel_executions = count
+
+    def update_last_data_message_time(self):
+        logging_info("Backpressure: Update last data message time")
+        """Update the last data message time"""
+        self.last_data_message_time = time.time()
+
+    def start_backpressure_monitor(self) -> Thread:
+        # Start the Backpressure monitor loop
+        self.thread = Thread(target=self.backpressure_monitor_loop)
+        self.thread.start()
+        return self.thread
+
+    async def check_overload_condition(self):
+        """Check if the current parallel executions exceed the limit"""
+        self.update_last_data_message_time()
+        logging_info(
+            "Backpressure: Check overload condition, current=%s, max=%s",
+            self.current_parallel_executions,
+            self.parallel_workers,
+        )
+        if self.current_parallel_executions >= self.parallel_workers - 1:
+            # Check if the last backpressure event was not an overload
+            if 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 = ScalingRequestAlert.OVERLOAD
+
+    def backpressure_monitor_loop(self):
+        _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: 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
+                    )
+                ):
+                    # 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,
+                        )
+                        # 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)
+
+        _loop.run_until_complete(_backpressure_monitor())
 
     async def handle_backpressure_overload_event(self):
         logging_warning("Backpressure: Capacity close to depleted!")
@@ -84,7 +260,7 @@ class BackpressureHandler:
         # update / reset time-window so that the OVERLOAD is not sent too often
         self.last_data_message_time = time.time()
 
-    def _handle_backpressure_idle_event(self):
+    async def _handle_backpressure_idle_event(self):
         logging_warning("Backpressure: Service is idle.")
         # Send an Idle event to the Management service
         scaling_request: ScaleRequestV1 = ScaleRequestV1(
@@ -95,26 +271,28 @@ class BackpressureHandler:
             ScalingRequestAlert.IDLE,
         )
         # Address the message to any adapter capable of supporting BACKPRESSURE request
-        self.publish_backpressure_request(scaling_request)
+        await self.publish_backpressure_request(scaling_request)
         # update / reset time-window so that the IDLE is not sent too often
         self.last_data_message_time = time.time()
 
     async def publish_backpressure_request(self, scaling_request: ScaleRequestV1):
         # Publish the backpressure request to the management service
+        logging_info(
+            f"Publishing backpressure for {scaling_request.serviceId} with request type {scaling_request.requestType}"
+        )
         if not self.exchange:
 
             async def _wrap_rabbit_mq_api_init(channel):
-                _exchange = await channel.declare_exchange(
-                    name="cleverthis.clevermicro.management", type=ExchangeType.fanout
+                _exchange = await channel.get_exchange(
+                    name="cleverthis.clevermicro.management"
                 )
                 return _exchange
 
-            _future = asyncio.run_coroutine_threadsafe(
-                _wrap_rabbit_mq_api_init(self.channel), self.loop
+            self.exchange = await await_result(
+                asyncio.run_coroutine_threadsafe(
+                    _wrap_rabbit_mq_api_init(self.channel), self.loop
+                )
             )
-            while not _future.done():
-                await asyncio.sleep(0.05)
-            self.exchange = _future.result()
 
         if self.exchange:
 
@@ -130,7 +308,9 @@ class BackpressureHandler:
                         priority=0,
                         correlation_id=None,
                     )
-                    await self.exchange.publish(message=pika_message, routing_key="")
+                    await self.exchange.publish(
+                        message=pika_message, routing_key="backpressure-scaling-v1"
+                    )
                     logging_info(
                         "Service Message Published to %s, msg: %s",
                         self.exchange.name,
@@ -139,6 +319,6 @@ class BackpressureHandler:
                     return True
                 return False
 
-            _future = asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop)
-            while not _future.done():
-                await asyncio.sleep(0.05)
+            await await_future(
+                asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop)
+            )

amqp/adapter/cleverthis_service_adapter.py

@@ -6,7 +6,6 @@ Methods here will invoke the actual CleverThis Service code, which are passed as
 import asyncio
 import inspect
 import json
-import logging
 from asyncio import AbstractEventLoop
 from typing import Dict, List, Callable, Coroutine, Any
 
@@ -27,6 +26,7 @@ from amqp.adapter.logging_utils import logging_error, logging_debug, logging_inf
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import DataMessage, DataResponse
 from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
+from amqp.router.utils import await_result
 
 
 @dataclass
@@ -86,12 +86,10 @@ async def _convert_file_response(
 
     # All calls to RabbitMQ API have to be called threadsafe, because those appear to be attached to parent
     # event loop, and need to execute at that event loop, not the current one
-    _future = asyncio.run_coroutine_threadsafe(_wrap_amq_api_calls(connection), loop)
-    while not _future.done():
-        await asyncio.sleep(
-            0.1
-        )  # allow coroutine to execute. calling _future.result() will block the entire thread and the coroutine will never execute
-    _channel, _exchange, _routing_key = _future.result()
+    # allow coroutine to execute. calling _future.result() will block the entire thread and the coroutine will never execute
+    _channel, _exchange, _routing_key = await await_result(
+        asyncio.run_coroutine_threadsafe(_wrap_amq_api_calls(connection), loop)
+    )
     # publish_file also needs RabbitMQ API, ensure the loop is passed as argument
     (_stream_name, _file_size) = await file_handler.publish_file(
         _exchange, obj.path, _routing_key, loop
@@ -142,23 +140,9 @@ class CleverThisServiceAdapter:
         self.service_host = self.amq_configuration.dispatch.service_host
         self.service_port = self.amq_configuration.dispatch.service_port
         self.loop: AbstractEventLoop | None = None
-        try:
-            self.require_authenticated_user = (
-                self.amq_configuration.amq_adapter.require_authenticated_user
-                and isinstance(
-                    self.amq_configuration.amq_adapter.require_authenticated_user, bool
-                )
-                or eval(
-                    str(self.amq_configuration.amq_adapter.require_authenticated_user)
-                    .strip()
-                    .capitalize()
-                    or "True"
-                )
-            )
-        except Exception as e:
-            self.require_authenticated_user = (
-                True  # Enforce authenticated user as default
-            )
+        self.require_authenticated_user = (
+            self.amq_configuration.amq_adapter.require_authenticated_user
+        )
 
     def get_content_type(self, message_headers: Dict[str, List[str]]) -> str:
         """

amqp/adapter/file_handler.py

@@ -20,6 +20,7 @@ from aio_pika.abc import (
 )
 
 from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
+from amqp.router.utils import await_result
 
 # Global dictionary to track completed downloads
 download_locations = defaultdict(str)
@@ -211,12 +212,9 @@ class StreamingFileHandler(FileHandler):
             return _file_size, _eof
 
         if queue_name:
-            _future: Future = asyncio.run_coroutine_threadsafe(
-                wrap_rabbit_mq_calls(), loop=loop
+            _local_file_size, _local_eof = await await_result(
+                asyncio.run_coroutine_threadsafe(wrap_rabbit_mq_calls(), loop=loop)
             )
-            while not _future.done():
-                await asyncio.sleep(0.05)
-            _local_file_size, _local_eof = _future.result()
             return _local_file_size, _local_eof
 
         return 0, CANCELLED
@@ -259,12 +257,9 @@ class StreamingFileHandler(FileHandler):
                 await _connection.close()
                 return _res, _eof
 
-            _future: Future = asyncio.run_coroutine_threadsafe(
-                wrap_rabbit_mq_calls(), loop=loop
+            _local_res, _local_eof = await await_result(
+                asyncio.run_coroutine_threadsafe(wrap_rabbit_mq_calls(), loop=loop)
             )
-            while not _future.done():
-                await asyncio.sleep(0.05)
-            _local_res, _local_eof = _future.result()
             return _local_res, _local_eof
         return 0, CANCELLED
 
@@ -399,12 +394,12 @@ class StreamingFileHandler(FileHandler):
                 while not _future.done():
                     # sleep allows coroutine to execute. must wait for completion to ensure all chunks written in right order
                     await asyncio.sleep(0.02)
-                    logging.debug(
+                    logging_debug(
                         f"Waiting for chunk {_offset}/{_file_size} future= {_future}"
                     )
 
                 _offset += len(_chunk)
-                logging.debug(
+                logging_debug(
                     f"Published chunk {_offset}/{_file_size} to {exchange.name}:{routing_key}"
                 )
 

amqp/adapter/logging_utils.py

@@ -2,6 +2,7 @@ import inspect
 import logging
 import os
 import sys
+import threading
 import traceback
 
 verbose = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1"
@@ -46,21 +47,22 @@ def logging_error(msg: str, *args) -> None:
         msg (str): The error message to log.
     """
     _exec_info = sys.exc_info()[2]
+    _thread_id = threading.get_ident()
     if _exec_info:
         _tb = traceback.extract_tb(sys.exc_info()[2])[-1]
         logging.error(
-            f" [E] '{msg}' ---> '{_tb.filename}':{_tb.lineno}, '{_tb.name}()'",
+            f"{_thread_id} [E] '{msg}' ---> '{_tb.filename}':{_tb.lineno}, '{_tb.name}()'",
             *args,
         )
     else:
         module, line_number, function_name = get_context_info()
         if module and line_number and function_name:
             logging.error(
-                f" [E] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+                f"{_thread_id} [E] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
                 *args,
             )
         else:
-            logging.error(f" [E] {msg}", *args)
+            logging.error(f"{_thread_id} [E] {msg}", *args)
 
 
 def logging_warning(msg: str, *args) -> None:
@@ -70,17 +72,18 @@ def logging_warning(msg: str, *args) -> None:
     Args:
         msg (str): The warning message to log.
     """
+    _thread_id = threading.get_ident()
     if not verbose:
-        logging.warning(f" [W] {msg}", *args)
+        logging.warning(f"{_thread_id} [W] {msg}", *args)
         return
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
         logging.warning(
-            f" [W] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id} [W] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.warning(f" [W] {msg}", *args)
+        logging.warning(f"{_thread_id} [W] {msg}", *args)
 
 
 def logging_info(msg: str, *args) -> None:
@@ -90,17 +93,18 @@ def logging_info(msg: str, *args) -> None:
     Args:
         msg (str): The info message to log.
     """
+    _thread_id = threading.get_ident()
     if not verbose:
-        logging.info(f" [*] {msg}", *args)
+        logging.info(f"{_thread_id} [*] {msg}", *args)
         return
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
         logging.info(
-            f" [*] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id} [*] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.info(f" [*] {msg}", *args)
+        logging.info(f"{_thread_id} [*] {msg}", *args)
 
 
 def logging_debug(msg: str, *args) -> None:
@@ -110,11 +114,12 @@ def logging_debug(msg: str, *args) -> None:
     Args:
         msg (str): The debug message to log.
     """
+    _thread_id = threading.get_ident()
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
         logging.info(
-            f" [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id} [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.debug(f" [DBG] {msg}", *args)
+        logging.debug(f"{_thread_id} [DBG] {msg}", *args)

amqp/config/amq_configuration.py

@@ -1,6 +1,5 @@
 import configparser
 import inspect
-import logging
 import os
 
 from amqp.adapter.logging_utils import logging_error, logging_warning
@@ -36,11 +35,16 @@ class AMQAdapter:
         self.route_mapping = config.get(
             "CleverMicro-AMQ", self.PREFIX + "route-mapping", fallback=None
         )
-        self.require_authenticated_user = config.get(
+        self.require_authenticated_user = config.getboolean(
             "CleverMicro-AMQ",
             self.PREFIX + "require-authenticated-user",
             fallback=False,
         )
+        # these 2 shall be provided by Docker swarm, format is a random alphanumeric string
+        self.swarm_service_id = os.getenv(
+            "SWARM_SERVICE_ID", default="clever-amqp-service"
+        )
+        self.swarm_task_id = os.getenv("SWARM_TASK_ID", default="python-adapter")
 
     def adapter_prefix(self) -> str:
         return f"{self.service_name}-{self.generator_id}-"
@@ -56,7 +60,6 @@ class Dispatch:
     def __init__(self, config: configparser.ConfigParser):
         """
         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
@@ -68,9 +71,6 @@ class Dispatch:
         self.use_dlq = config.getboolean(
             "CleverMicro-AMQ", self.PREFIX + "use-dlq", fallback=False
         )
-        self.use_confirms = config.getboolean(
-            "CleverMicro-AMQ", self.PREFIX + "use-confirms", fallback=False
-        )
         self.amq_host = config.get(
             "CleverMicro-AMQ", self.PREFIX + "amq-host", fallback="rabbitmq"
         )
@@ -110,12 +110,27 @@ class Backpressure:
 
         :param config: config parser to read configuration values
         """
-        self.threshold = config.getint(
-            "CleverMicro-AMQ", self.PREFIX + "threshold", fallback=0
+        self.threshold_threads = config.getint(
+            "CleverMicro-AMQ",
+            self.PREFIX + "threshold",
+            fallback=int(os.getenv("PARALLEL_WORKERS", 1)),
         )
-        # time-window is in milliseconds, to report backpressure IDLE alert
+        # time-window in seconds, duration between backpressure reports
         self.time_window = config.getint(
-            "CleverMicro-AMQ", self.PREFIX + "time-window", fallback=10000
+            "CleverMicro-AMQ", self.PREFIX + "time-window", fallback=10
+        )
+
+        # Define maximum CPU usage before triggering backpressure OVERLOAD alert
+        self.threshold_cpu = config.getint(
+            "CleverMicro-AMQ", self.PREFIX + "threshold-cpu", fallback=90
+        )
+        # Define the time window length, in seconds, during which, if CPU usage is consistently above the threshold, the system will trigger OVERLOAD alert
+        self.cpu_overload_duration = config.getint(
+            "CleverMicro-AMQ", self.PREFIX + "cpu-overload-duration", fallback=30
+        )
+        # Define the time window length, in seconds, during which, if no request is made to the Service, the system will trigger IDLE alert
+        self.idle_duration = config.getint(
+            "CleverMicro-AMQ", self.PREFIX + "idle-duration", fallback=30
         )
 
 

amqp/config/application.properties

@@ -2,21 +2,41 @@
 # CleverMicro AMQ Adapter settings
 # ====================================================================
 [CleverMicro-AMQ]
-cm.amq-adapter.service-name=amq-adapter-python
-cm.amq-adapter.generator-id=164
-cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:cleverswarm.localhost.#:5:0
+# CleverMicro AMQ Adapter settings that identify this AMQ Adapter instance to other adapters
+
+# 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=20
+cm.backpressure.threshold=5
+# Define maximum CPU usage (%) before triggering backpressure OVERLOAD alert
+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.cpu-idle-duration=30
+

amqp/router/utils.py

@@ -1,6 +1,8 @@
+import asyncio
 import re
 
 pattern = re.compile(r"[^a-zA-Z0-9.#/\-]")
+AWAIT_SLEEP = 0.05  # seconds
 
 
 def sanitize_as_rabbitmq_name(input_str: str) -> str:
@@ -17,3 +19,19 @@ def sanitize_as_rabbitmq_domain_name(token: str) -> str:
     # Step 3: Remove trailing '.' character, if applicable
     step_last_char = len(step2) - 1
     return step2[:-1] if step_last_char > 0 and step2[step_last_char] == "." else step2
+
+
+async def await_future(future: asyncio.Future, sleep_time=AWAIT_SLEEP):
+    """
+    Wait for a future to complete.
+    """
+    while not future.done():
+        await asyncio.sleep(sleep_time)
+
+
+async def await_result(future: asyncio.Future, sleep_time=AWAIT_SLEEP):
+    """
+    Wait for a future to complete and return its result.
+    """
+    await await_future(future, sleep_time=sleep_time)
+    return future.result()

amqp/service/amq_message_handler.py

@@ -1,8 +1,6 @@
 import asyncio
-import logging
 import threading
 import time
-import concurrent.futures
 import os
 from asyncio import Future, AbstractEventLoop
 from typing import Dict
@@ -30,11 +28,6 @@ from amqp.model.model import (
 from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
 from amqp.adapter.serializer import map_as_string
 
-# Shared thread pool for parallel execution
-executor = concurrent.futures.ThreadPoolExecutor(
-    max_workers=BackpressureHandler.parallel_workers + 2
-)
-
 
 def collect_trace_info():
     trace_map = {}
@@ -63,6 +56,7 @@ class DataMessageHandler:
         rabbit_mq_client: RabbitMQClient,
         amq_configuration: AMQConfiguration,
         loop,
+        backpressure_handler: BackpressureHandler,
         file_handler: FileHandler = StreamingFileHandler(),
     ):
         self.service_adapter: CleverThisServiceAdapter = service_adapter
@@ -78,29 +72,24 @@ class DataMessageHandler:
         # Dictionary to store outstanding Futures
         self.outstanding: Dict[str, asyncio.Future] = {}
         self.reply_to: str | None = None
-        self.backpressure_monitor_window = (
-            amq_configuration.backpressure.time_window / 1000
-        )  # convert to seconds
         self.file_handler: FileHandler = file_handler
-        self.backpressure_handler = BackpressureHandler(
-            channel=rabbit_mq_client.channel, loop=loop
-        )
-        _backpressure_treshold = amq_configuration.backpressure.threshold
-        """
-        Determine the number of parallel workers based on the backpressure threshold and PARALLEL_WORKERS env var.
-        If env var PARALLEL_WORKERS is not set (or set to 0), use the backpressure threshold config
-        as the number of parallel workers.
-        """
-        if _backpressure_treshold > 0:
-            if BackpressureHandler.parallel_workers <= 0:
-                BackpressureHandler.parallel_workers = _backpressure_treshold
+        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.
@@ -143,19 +132,15 @@ class DataMessageHandler:
         :return: DataResponse
         """
         # Increment the counter for parallel executions
-        BackpressureHandler.current_parallel_executions += 1
+        self.backpressure_handler.increase_parallel_executions()
         logging_info(
-            f"PARALLEL: Current Capacity [{BackpressureHandler.current_parallel_executions}] of [{BackpressureHandler.parallel_workers}]"
+            f"PARALLEL: Current Capacity [{self.backpressure_handler.current_parallel_executions}] of [{self.backpressure_handler.parallel_workers}]"
         )
         if message.reply_to:
             self.reply_to = message.reply_to
-        if (
-            BackpressureHandler.current_parallel_executions
-            > BackpressureHandler.parallel_workers - 2
-        ):
-            await self.backpressure_handler.handle_backpressure_overload_event()
 
         try:
+            await self.backpressure_handler.check_overload_condition()
             _a_message = AMQMessage(amq_message, self.rabbit_mq_client.connection)
             _response: DataResponse = await self.service_adapter.on_message(_a_message)
             logging_debug(f"Result in thread: {_response}")
@@ -175,11 +160,10 @@ class DataMessageHandler:
                         os.remove(filename)
                     except OSError as e:
                         logging_error(f"Deleting file {filename}: {e}")
-            BackpressureHandler.current_parallel_executions -= 1
-            return
         except Exception as e:
             logging_error(f"Request handler: {e}")
-            return
+
+        self.backpressure_handler.decrease_parallel_executions()
 
     async def reconstitute_data_message(
         self, message: AbstractIncomingMessage

amqp/service/amq_service.py

@@ -3,9 +3,12 @@ import logging.config
 import os
 import asyncio
 from asyncio import Future
+from threading import Thread
+from typing import Optional
 
 from aio_pika.abc import AbstractRobustExchange
 
+from amqp.adapter.backpressure_handler import BackpressureHandler
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.adapter.logging_utils import logging_info, logging_warning
@@ -33,13 +36,14 @@ class AMQService:
     CLEVER_AMQP_CLIENT_VERSION = os.getenv("CLEVER_AMQP_CLIENT_VERSION")
     MAX_WAIT = 300  # defaults to 5 minutes (300 seconds)
 
-    def __init__(self, amq_configuration: AMQConfiguration, service_adapter):
+    def __init__(self, amq_configuration: AMQConfiguration, service_adapter=None):
         logging.info("***********************************************")
         logging.info("**********  AMQ   Service   *******************")
         logging.info("***********************************************")
-        self.loop = asyncio.get_event_loop()
+        self.loop = asyncio.new_event_loop()
 
         self.amq_configuration = amq_configuration
+        self.service_adapter = service_adapter
         self.data_message_factory = DataMessageFactory.get_instance(
             self.amq_configuration.amq_adapter.generator_id
         )
@@ -49,8 +53,12 @@ class AMQService:
         self.service_message_factory = ServiceMessageFactory(
             self.amq_configuration.amq_adapter.service_name
         )
+        self.backpressure_thread: Optional[Thread] = None
+
+    def run(self):
         # self.init(loop) will initialize RabbitMQ connection
-        self.loop.run_until_complete(self.init(self.loop, service_adapter))
+        asyncio.set_event_loop(self.loop)
+        self.loop.run_until_complete(self.init(self.loop, self.service_adapter))
 
         logging.info("***********************************************")
         logging.info("******  AMQ   Service  Init  Complete *********")
@@ -61,6 +69,11 @@ class AMQService:
         _reply_to_exchange: AbstractRobustExchange = await self.rabbit_mq_client.init(
             loop
         )
+        backpressure_handler = BackpressureHandler(
+            channel=self.rabbit_mq_client.channel,
+            loop=loop,
+            config=self.amq_configuration,
+        )
         self.amq_data_message_handler = DataMessageHandler(
             service_adapter,
             self.tracer,
@@ -70,9 +83,11 @@ class AMQService:
             self.rabbit_mq_client,
             self.amq_configuration,
             loop=loop,
+            backpressure_handler=backpressure_handler,
         )
         await self.register_routes()
         await self.rabbit_mq_client.request_routes()
+        self.backpressure_thread = backpressure_handler.start_backpressure_monitor()
 
     def cleanup(self):
         logging_info("AMQService.cleanup() rmqc=%s", self.rabbit_mq_client)
@@ -92,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

@@ -8,6 +8,7 @@ import importlib
 import pathlib
 import re
 from collections import defaultdict
+from threading import Thread
 from typing import Optional, List, Dict
 from aiohttp import ClientSession
 from fastapi.routing import APIRoute
@@ -128,7 +129,8 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
         super().__init__(amq_configuration, session)
         self.routes = group_routes_by_method(api_routes=api.api_router.routes)
         _amq_service: AMQService = AMQService(amq_configuration, self)
-        _amq_service.rabbit_mq_client.channel.start_consuming()
+        self.thread = Thread(target=_amq_service.run)
+        self.thread.start()
 
     async def get_current_user(self, token: str) -> UserSchema:
         """
@@ -250,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(

cleverswarm/main.py

@@ -1,14 +1,8 @@
-import getch
 from amqp.config.amq_configuration import AMQConfiguration
 
 from clever_swarm_adapter import CleverSwarmAmqpAdapter
 
 if __name__ == "__main__":
     adapter = CleverSwarmAmqpAdapter(AMQConfiguration("./application.properties.local"))
-
-    print("Press the space bar to exit...")
-    while True:
-        key = getch.getch()  # Wait for a keypress
-        if key == " ":  # Check if the key is the space bar
-            print("Space bar pressed. Exiting...")
-            break
+    print("Press ^C to exit...")
+    adapter.thread.join()

pyproject.toml

@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.21-dev"
-description = "The Python implementation of the AMQ Adaptor for CleverMicro"
+version = "0.2.27"
+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.115.6",
+    "fastapi==0.114.2",
     "opentelemetry-api",
     "opentelemetry-sdk",
     "opentelemetry-exporter-otlp",
     "opentelemetry-exporter-prometheus",
     "opentelemetry-instrumentation-pika",
+    "pika~=1.3.2",
+    "psutil",
     "python_multipart"
 ]
 

tests/adapter/test_cleverthis_service_adapter.py

@@ -222,25 +222,6 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
         with self.assertRaises(NotImplementedError):
             await self.adapter.delete(mock_message, None)
 
-    def test_require_authenticated_user_config(self):
-        # Test different config values for require_authenticated_user
-        test_cases = [
-            ("True", True),
-            ("False", False),
-            ("", True),  # Default when empty
-            ("invalid", True),  # Default when invalid
-        ]
-
-        for config_value, expected in test_cases:
-            with self.subTest(config_value=config_value):
-                mock_config = AMQConfiguration("./application.properties.local")
-                mock_config.dispatch.service_host = "testhost"
-                mock_config.dispatch.service_port = 8080
-                mock_config.amq_adapter.require_authenticated_user = config_value
-
-                adapter = CleverThisServiceAdapter(mock_config)
-                self.assertEqual(adapter.require_authenticated_user, expected)
-
 
 if __name__ == "__main__":
     unittest.main()

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()