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

amqp/adapter/backpressure_handler.py

@@ -4,7 +4,6 @@ import time
 from asyncio import AbstractEventLoop
 from threading import Thread
 
-import psutil
 from aio_pika import Message
 from aio_pika.abc import AbstractRobustChannel
 
@@ -47,40 +46,6 @@ 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
@@ -104,7 +69,6 @@ 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"""
@@ -155,93 +119,67 @@ class BackpressureHandler:
                 self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
 
     def backpressure_monitor_loop(self):
-        _time_window_millis = self.config.backpressure.time_window
+        _time_window_sec = self.config.backpressure.time_window
-        _idle_duration_millis = self.config.backpressure.idle_duration
+        _idle_duration_millis = 1000 * 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",
+                    "Backpressure: Monitor loop, current=%s",
-                    _current_cpu_usage,
+                    self.current_parallel_executions,
-                    _average_cpu_usage,
+                )
-                    _cpu_usage_state,
+                logging_info(
+                    "Backpressure: Last data message time=%s, eventTime=%s",
                     self.last_data_message_time,
-                    self.last_backpressure_event,
                     self.last_backpressure_event_time,
                 )
-                if _average_cpu_usage < IDLE_THRESHOLD:
+                logging_info(
-                    if _cpu_usage_state != CPU_IDLE:
+                    "Backpressure: Last backpressure event=%s",
-                        _cpu_usage_changed = _now
+                    self.last_backpressure_event,
-                    _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
+                    self.current_parallel_executions >= self.parallel_workers
-                    and _now - _cpu_usage_changed > _overload_duration_millis
+                    and time.time() - self.last_backpressure_event_time
-                    and (
+                    > _time_window_sec
-                        _now - self.last_backpressure_event_time > _time_window_millis
+                    and self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
-                        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_time = time.time()
                     self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
                 elif (
-                    _cpu_usage_state == CPU_IDLE
+                    self.current_parallel_executions == 0
-                    and _now - _cpu_usage_changed > _idle_duration_millis
+                    and time.time() - self.last_data_message_time
-                    and _now - self.last_data_message_time > _idle_duration_millis
+                    >= _idle_duration_millis
-                    and (
+                    and self.last_backpressure_event != ScalingRequestAlert.IDLE
-                        _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
+                    self.last_backpressure_event_time = time.time()
                 else:
                     # Trigger update event
-                    if _now - self.last_backpressure_event_time > _time_window_millis:
+                    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,
-                            100,
+                            self.parallel_workers,
-                            _average_cpu_usage,
+                            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 = _now
+                        self.last_backpressure_event_time = time.time()
                         self.last_backpressure_event = ScalingRequestAlert.UPDATE
-                await asyncio.sleep(_monitor_interval)
+                await asyncio.sleep(_time_window_sec)
 
         _loop.run_until_complete(_backpressure_monitor())
 

amqp/config/application.properties

@@ -2,41 +2,30 @@
 # CleverMicro AMQ Adapter settings
 # ====================================================================
 [CleverMicro-AMQ]
-# CleverMicro AMQ Adapter settings that identify this AMQ Adapter instance to other adapters
+cm.amq-adapter.service-name=amq-adapter-python
-
+cm.amq-adapter.generator-id=164
-# A name to identify this service for the AMQ Adapter
+cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:cleverswarm.localhost.#:5:0
-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 CleveSwarm nor CleverBRAG or in tight coupling mode.
+# Not used for CleverSwarm 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=cleverthis-service-name.localhost
+#cm.dispatch.service-host=cleverswarm.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-overload=90
+cm.backpressure.threshold-cpu=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
+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 aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange, ExchangeType
-from pika.exchange_type import 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()
+                cm_message.is_valid()
-            and not self.amq_configuration.amq_adapter.service_name
+                and not self.amq_configuration.amq_adapter.service_name
-            == cm_message.reply_to
+                        == 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/service/amq_message_handler.py

@@ -75,21 +75,12 @@ class DataMessageHandler:
         self.file_handler: FileHandler = file_handler
         self.backpressure_handler: BackpressureHandler = backpressure_handler
 
-    async def inbound_data_message_callback_as_thread(
+    async def inbound_data_message_callback(self, message: AbstractIncomingMessage):
-        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(
+            target=lambda: asyncio.run(self.run_callback_thread(message))
-                self.inbound_data_message_callback_no_thread(message)
-            )
         ).start()
 
-    async def inbound_data_message_callback_no_thread(
+    async def run_callback_thread(self, message: AbstractIncomingMessage):
-        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_no_thread,
+                    self.amq_data_message_handler.inbound_data_message_callback,
                 )
                 return await self.rabbit_mq_client.register_data_reply_callback(
                     self.amq_data_message_handler.reply_received_callback

cleverswarm/clever_swarm_adapter.py

@@ -252,13 +252,11 @@ 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(
+                                file=pathlib.Path(actual_filepath).open("rb"),
-                                    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.27"
+version = "0.2.23-dev"
-description = "The Python implementation of the AMQ Adapter for CleverMicro"
+description = "The Python implementation of the AMQ Adaptor for CleverMicro"
 readme = "README.md"
 authors = [
     { name = "Stanislav Hejny", email = "stanislav.hejny@cleverthis.com" }
@@ -19,16 +19,15 @@ classifiers = [
 ]
 requires-python = ">=3.11"
 dependencies = [
-    "aio-pika~=9.5.5",
+    "aio-pika~=9.5",
-    "aiohttp~=3.11.11",
+    "aiohttp~=3.11",
-    "fastapi==0.114.2",
+    "aio_pika~=9.5",
+    "fastapi~=0.114",
     "opentelemetry-api",
     "opentelemetry-sdk",
     "opentelemetry-exporter-otlp",
     "opentelemetry-exporter-prometheus",
     "opentelemetry-instrumentation-pika",
-    "pika~=1.3.2",
-    "psutil",
     "python_multipart"
 ]
 

tools/create_exchange.py

@@ -1,90 +0,0 @@
-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()