test

ISSUES CLOSED: clevermicro/amq-adapter-python#8

amqp/adapter/backpressure_handler.py

@@ -1,17 +1,23 @@
 import asyncio
 import json
+import logging
+import os
 import time
 from asyncio import AbstractEventLoop
 from threading import Thread
 
 from aio_pika import Message
-from aio_pika.abc import AbstractRobustChannel
+from aio_pika.abc import AbstractRobustChannel, ExchangeType
 
 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
 
+# Duration in seconds between backpressure reports
+REPORT_WINDOW_SEC = 10
+# Number of reports without activity to trigger idle event
+NUM_REPORTS_FOR_IDLE = 3
+
 
 class ScaleRequestV1:
 
@@ -55,20 +61,15 @@ class BackpressureHandler:
     last_backpressure_event_time = 0
     last_backpressure_event = ScalingRequestAlert.UPDATE
 
-    def __init__(
-        self,
-        channel: AbstractRobustChannel,
-        loop: AbstractEventLoop,
-        config: AMQConfiguration,
-    ):
+    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):
         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 increase_parallel_executions(self):
         """Increase the number of parallel executions"""
@@ -119,8 +120,7 @@ class BackpressureHandler:
                 self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
 
     def backpressure_monitor_loop(self):
-        _time_window_sec = self.config.backpressure.time_window
-        _idle_duration_millis = 1000 * self.config.backpressure.idle_duration
+        global REPORT_WINDOW_SEC
         _loop = asyncio.new_event_loop()
 
         async def _backpressure_monitor():
@@ -143,7 +143,7 @@ class BackpressureHandler:
                 if (
                     self.current_parallel_executions >= self.parallel_workers
                     and time.time() - self.last_backpressure_event_time
-                    > _time_window_sec
+                    > REPORT_WINDOW_SEC
                     and self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
                 ):
                     # Trigger the overload event
@@ -153,7 +153,7 @@ class BackpressureHandler:
                 elif (
                     self.current_parallel_executions == 0
                     and time.time() - self.last_data_message_time
-                    >= _idle_duration_millis
+                    > NUM_REPORTS_FOR_IDLE * REPORT_WINDOW_SEC
                     and self.last_backpressure_event != ScalingRequestAlert.IDLE
                 ):
                     # Trigger the idle event
@@ -165,7 +165,7 @@ class BackpressureHandler:
                     if (
                         self.current_parallel_executions > 0
                         and time.time() - self.last_backpressure_event_time
-                        > _time_window_sec
+                        > REPORT_WINDOW_SEC
                     ):
                         # Trigger the update event
                         _scaling_request: ScaleRequestV1 = ScaleRequestV1(
@@ -179,7 +179,7 @@ class BackpressureHandler:
                         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)
+                await asyncio.sleep(REPORT_WINDOW_SEC)
 
         _loop.run_until_complete(_backpressure_monitor())
 

amqp/adapter/cleverthis_service_adapter.py

@@ -6,6 +6,7 @@ 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
 
@@ -140,9 +141,23 @@ 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
-        self.require_authenticated_user = (
-            self.amq_configuration.amq_adapter.require_authenticated_user
-        )
+        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
+            )
 
     def get_content_type(self, message_headers: Dict[str, List[str]]) -> str:
         """

amqp/config/amq_configuration.py

@@ -1,5 +1,6 @@
 import configparser
 import inspect
+import logging
 import os
 
 from amqp.adapter.logging_utils import logging_error, logging_warning
@@ -35,16 +36,11 @@ class AMQAdapter:
         self.route_mapping = config.get(
             "CleverMicro-AMQ", self.PREFIX + "route-mapping", fallback=None
         )
-        self.require_authenticated_user = config.getboolean(
+        self.require_authenticated_user = config.get(
             "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}-"
@@ -60,6 +56,7 @@ 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
@@ -71,6 +68,9 @@ 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,27 +110,12 @@ class Backpressure:
 
         :param config: config parser to read configuration values
         """
-        self.threshold_threads = config.getint(
-            "CleverMicro-AMQ",
-            self.PREFIX + "threshold",
-            fallback=int(os.getenv("PARALLEL_WORKERS", 1)),
+        self.threshold = config.getint(
+            "CleverMicro-AMQ", self.PREFIX + "threshold", fallback=0
         )
-        # time-window in seconds, duration between backpressure reports
+        # time-window is in milliseconds, to report backpressure IDLE alert
         self.time_window = config.getint(
-            "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
+            "CleverMicro-AMQ", self.PREFIX + "time-window", fallback=10000
         )
 
 

amqp/config/application.properties

@@ -19,13 +19,4 @@ 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=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
-
+cm.backpressure.threshold=20

amqp/service/amq_message_handler.py

@@ -1,6 +1,8 @@
 import asyncio
+import logging
 import threading
 import time
+import concurrent.futures
 import os
 from asyncio import Future, AbstractEventLoop
 from typing import Dict
@@ -28,6 +30,11 @@ 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 = {}
@@ -72,8 +79,20 @@ 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 = backpressure_handler
+        _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
 
     async def inbound_data_message_callback(self, message: AbstractIncomingMessage):
         threading.Thread(

amqp/service/amq_service.py

@@ -70,9 +70,7 @@ class AMQService:
             loop
         )
         backpressure_handler = BackpressureHandler(
-            channel=self.rabbit_mq_client.channel,
-            loop=loop,
-            config=self.amq_configuration,
+            channel=self.rabbit_mq_client.channel, loop=loop
         )
         self.amq_data_message_handler = DataMessageHandler(
             service_adapter,

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.23-dev"
+version = "0.2.22-dev"
 description = "The Python implementation of the AMQ Adaptor for CleverMicro"
 readme = "README.md"
 authors = [