fix(General): Fix deps version so cleverswarm can install it

This MR fixed the pipeline so it can run on forgejo runner. Currently we have two pipelines set up:

+ One pipeline will run pytest to test the code and collect coverage data, this pipeline will run on all branches and PRs.
+ One pipeline will build the wheel file and publish it to the company's pypl repo, this pipeline will only run on the master branch and develop branch.

This RP also made the following changes to make the pipeline working:

+ Remove `setup.py` and `requirements.txt`, now the project is fully embraced `pyproject.toml`, which according to ChatGPT, is the most modern way of managing a python project.
+ Move all test files to `tests` folder, so it won't be packed into the final wheel file and affect the coverage rate.
+ Change version to `X.Y.Z-dev`, before building the wheel file, the pipeline will automatically add a UTC timestamp after it, so it becomes `X.Y.Z-devYYYYMMDDHHmmSS`. This is added due to the limitation of the pypl repo, unlike maven snapshot versions, you can't push the same version twice with our pypl repository.

Reviewed-on: #10
Reviewed-by: Stanislav Hejny <stanislav.hejny@cleverthis.com>
Co-authored-by: Rui Hu <rui.hu@cleverthis.com>
Co-committed-by: Rui Hu <rui.hu@cleverthis.com>

.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,16 @@
 import asyncio
 import json
-import logging
-import os
-import threading
 import time
 from asyncio import AbstractEventLoop
+from threading import Thread
 
 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 +29,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"""
@@ -48,26 +49,139 @@ class ScaleRequestV1:
 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
 
-    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_sec = self.config.backpressure.time_window
+        _idle_duration_millis = 1000 * self.config.backpressure.idle_duration
+        _loop = asyncio.new_event_loop()
+
+        async def _backpressure_monitor():
+            """Monitor the backpressure conditions and trigger events accordingly"""
+            while True:
+                logging_info(
+                    "Backpressure: Monitor loop, current=%s",
+                    self.current_parallel_executions,
+                )
+                logging_info(
+                    "Backpressure: Last data message time=%s, eventTime=%s",
+                    self.last_data_message_time,
+                    self.last_backpressure_event_time,
+                )
+                logging_info(
+                    "Backpressure: Last backpressure event=%s",
+                    self.last_backpressure_event,
+                )
+                # 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
+                ):
+                    # Trigger the overload event
+                    await self.handle_backpressure_overload_event()
+                    self.last_backpressure_event_time = time.time()
+                    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
+                ):
+                    # Trigger the idle event
+                    await self._handle_backpressure_idle_event()
+                    self.last_backpressure_event = ScalingRequestAlert.IDLE
+                    self.last_backpressure_event_time = time.time()
+                else:
+                    # Trigger update event
+                    if (
+                        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,
+                            self.parallel_workers - self.current_parallel_executions,
+                            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 = ScalingRequestAlert.UPDATE
+                await asyncio.sleep(_time_window_sec)
+
+        _loop.run_until_complete(_backpressure_monitor())
 
     async def handle_backpressure_overload_event(self):
         logging_warning("Backpressure: Capacity close to depleted!")
@@ -84,7 +198,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 +209,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 +246,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 +257,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

@@ -19,4 +19,13 @@ cm.dispatch.download-dir=/tmp/downloaded_files
 #cm.dispatch.service-port=8080
 #
 # 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=90
+# 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
+

amqp/rabbitmq/rabbit_mq_client.py

@@ -1,8 +1,7 @@
 import logging
 
 import aio_pika
-from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange
-from pika.exchange_type import ExchangeType
+from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange, ExchangeType
 
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.logging_utils import (
@@ -55,7 +54,7 @@ class RabbitMQClient:
             name=cm_reply_to_queue, durable=True, exclusive=False, auto_delete=False
         )
         self.reply_to_exchange = await self.channel.declare_exchange(
-            AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.direct
+            AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.DIRECT
         )
         await self.reply_to_queue.bind(
             exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue
@@ -114,7 +113,7 @@ class RabbitMQClient:
                     )
                     exchange: AbstractRobustExchange = (
                         await self.channel.declare_exchange(
-                            name=_exchange_name, type=ExchangeType.topic
+                            name=_exchange_name, type=ExchangeType.TOPIC
                         )
                     )
                     await queue.bind(exchange, route.key)

amqp/router/router_producer.py

@@ -15,9 +15,8 @@ from aio_pika.abc import (
     AbstractRobustQueue,
     AbstractRobustConnection,
     AbstractRobustExchange,
-    AbstractIncomingMessage,
+    AbstractIncomingMessage, ExchangeType,
 )
-from pika.exchange_type import ExchangeType
 
 from amqp.adapter.logging_utils import (
     logging_error,
@@ -82,13 +81,13 @@ class RouterProducer(RouterBase):
                 # **** set up the service discovery internal exchanges ****
                 # 1. declare RequestExchange to where to publish the RoutingData requests
                 self.cm_request_exchange = await self.channel.declare_exchange(
-                    name=CM_REQUEST_EXCHANGE, type=ExchangeType.fanout
+                    name=CM_REQUEST_EXCHANGE, type=ExchangeType.FANOUT
                 )
                 #
                 # 2. declare ResponseExchange where to publish current RoutingData (response to
                 #    a RoutingData request)
                 self.cm_response_exchange = await self.channel.declare_exchange(
-                    name=CM_RESPONSE_EXCHANGE, type=ExchangeType.fanout
+                    name=CM_RESPONSE_EXCHANGE, type=ExchangeType.FANOUT
                 )
                 # 2a. declare a Response queue for Routing Data replies from Services
                 _cm_response_queue: str = self.get_response_queue_name()
@@ -145,9 +144,9 @@ class RouterProducer(RouterBase):
             cm_message.reply_to,
         )
         if (
-            cm_message.is_valid()
-            and not self.amq_configuration.amq_adapter.service_name
-            == cm_message.reply_to
+                cm_message.is_valid()
+                and not self.amq_configuration.amq_adapter.service_name
+                        == cm_message.reply_to
         ):
             if cm_message.message_type == ServiceMessageType.ROUTING_DATA:
                 self.add_routes(cm_message.message)
@@ -167,7 +166,7 @@ class RouterProducer(RouterBase):
                 ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT
             )
             if not await self.publish_service_message(
-                serviceMessage, self.cm_request_exchange
+                    serviceMessage, self.cm_request_exchange
             ):
                 logging_warning(
                     "RouterProducer could not request route data: %s channel is not open.",
@@ -178,7 +177,7 @@ class RouterProducer(RouterBase):
             logging_error("RouterProducer failed request routing data: %s", ioe)
 
     async def publish_service_message(
-        self, message: ServiceMessage, exchange: AbstractRobustExchange
+            self, message: ServiceMessage, exchange: AbstractRobustExchange
     ) -> bool:
         """
         * Publishes a service message to the service queue and logs success log entry.

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,22 +72,8 @@ 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):
         threading.Thread(
@@ -143,19 +123,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 +151,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)

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:
         """

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,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.21-dev"
+version = "0.2.23-dev"
 description = "The Python implementation of the AMQ Adaptor for CleverMicro"
 readme = "README.md"
 authors = [
@@ -19,10 +19,10 @@ classifiers = [
 ]
 requires-python = ">=3.11"
 dependencies = [
-    "aio-pika~=9.5.5",
-    "aiohttp~=3.11.11",
-    "aio_pika~=9.5.5",
-    "fastapi==0.115.6",
+    "aio-pika~=9.5",
+    "aiohttp~=3.11",
+    "aio_pika~=9.5",
+    "fastapi~=0.114",
     "opentelemetry-api",
     "opentelemetry-sdk",
     "opentelemetry-exporter-otlp",

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