diff --git a/amqp/adapter/data_message_factory.py b/amqp/adapter/data_message_factory.py index ed6ff15..040b79d 100644 --- a/amqp/adapter/data_message_factory.py +++ b/amqp/adapter/data_message_factory.py @@ -113,6 +113,25 @@ class DataMessageFactory: payload, ) + def safe_rsplit(self, fq_method_name, default="_"): + """ + Safely split a fully qualified method name into package, class, and method + + Args: + fq_method_name (str): Fully qualified method name + default (str, optional): Default value for missing tokens. Defaults to ''. + + Returns: + tuple: (package, class_name, method) + """ + if not fq_method_name: + return default, default, default + parts = fq_method_name.rsplit(".", 2) + # Pad with default values if insufficient tokens + while len(parts) < 3: + parts.insert(0, default) + return tuple(parts) + def create_rpc_message( self, fq_method_name: str, @@ -125,7 +144,7 @@ class DataMessageFactory: :param args: arguments whose serialized form will form the request body :return: DataMessage object """ - package, class_name, method = fq_method_name.rsplit(".", 2) + package, class_name, method = self.safe_rsplit(fq_method_name) headers = {"Content-Type": ["application/octet-stream"], "Return-Address": [swarm_id]} trace_info = {} body = python_type_to_json(args).encode(encoding="utf-8") if args else b"" diff --git a/amqp/config/amq_configuration.py b/amqp/config/amq_configuration.py index 79dd4e3..f6f598e 100644 --- a/amqp/config/amq_configuration.py +++ b/amqp/config/amq_configuration.py @@ -43,6 +43,11 @@ class AMQAdapter: self.PREFIX + "default-backpressure-enabled", fallback=False, ) + self.callbacks_enabled = config.getboolean( + "CleverMicro-AMQ", + self.PREFIX + "callbacks-enabled", + 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") diff --git a/amqp/config/application.properties b/amqp/config/application.properties index 7a0c3c7..b907530 100644 --- a/amqp/config/application.properties +++ b/amqp/config/application.properties @@ -20,6 +20,7 @@ cm.amq-adapter.consul-port=8500 cm.amq-adapter.consul-max-retries=5 cm.amq-adapter.consul-initial-retry-delay=0.2 cm.amq-adapter.consul-id-generator-key=services/ids/counter +cm.amq-adapter.callbacks-enabled=False # CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding cm.dispatch.use-dlq=true diff --git a/amqp/rabbitmq/rabbit_mq_client.py b/amqp/rabbitmq/rabbit_mq_client.py index d839ce6..720a4fe 100644 --- a/amqp/rabbitmq/rabbit_mq_client.py +++ b/amqp/rabbitmq/rabbit_mq_client.py @@ -1,4 +1,5 @@ import logging +from typing import Optional import aio_pika from aio_pika.abc import AbstractRobustExchange, AbstractRobustQueue @@ -95,19 +96,25 @@ class RabbitMQClient: route.queue if route.queue else self.router.get_consume_queue_name() ) logging_info("RabbitMQClient register route: [%s]", route) - queue: AbstractRobustQueue = await self.channel.declare_queue( - name=queue_name, - durable=True, - exclusive=False, - auto_delete=False, - arguments=queue_args, - ) + if self.amq_configuration.amq_adapter.callbacks_enabled: + queue: Optional[AbstractRobustQueue] = await self.channel.declare_queue( + name=queue_name, + durable=True, + exclusive=False, + auto_delete=False, + arguments=queue_args, + ) + else: + queue = None _exchange_name: str = route.exchange if route.exchange else AMQ_RPC_EXCHANGE exchange: AbstractRobustExchange = await self.channel.declare_exchange( name=_exchange_name, type=ExchangeType.topic ) - await queue.bind(exchange, route.key) - _c_tag = await queue.consume(callback, no_ack=False) + if self.amq_configuration.amq_adapter.callbacks_enabled: + await queue.bind(exchange, route.key) + _c_tag = await queue.consume(callback, no_ack=False) + else: + _c_tag = None # Register the route with the router and publish its detail to other adapters await self.router.register_route( AMQRoute( @@ -144,6 +151,9 @@ class RabbitMQClient: Set up the RabbitMQ connection :return: nothing, it applies changes to instance variables """ + logging.info( + f"RabbitMQClient.setupAMQ, connecting to RabbitMQ at {self.amq_configuration.dispatch.amq_host}:{self.amq_configuration.dispatch.amq_port}, loop={loop}" + ) self.connection = await aio_pika.connect_robust( host=self.amq_configuration.dispatch.amq_host, port=self.amq_configuration.dispatch.amq_port, diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index 2fd37e3..784b92e 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -30,6 +30,7 @@ from amqp.model.model import ( MultipartDataMessage, ) from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient +from amqp.router.utils import sanitize_as_rabbitmq_name def collect_trace_info(): @@ -470,7 +471,9 @@ class DataMessageHandler: name=rpc_exchange_name, type=ExchangeType.topic ) # Publish the message to the exchange with the component name as the routing key - await exchange.publish(message=pika_message, routing_key=route.component_name) + await exchange.publish( + message=pika_message, routing_key=sanitize_as_rabbitmq_name(route.component_name) + ) logging_debug( "###### RPC Request Published to exchg:(%s) routing_key=%s, msgId=%s", @@ -514,7 +517,9 @@ class DataMessageHandler: name=rpc_exchange_name, type=ExchangeType.topic ) # Publish the message to the exchange with the component name as the routing key - await exchange.publish(message=pika_message, routing_key=route.component_name) + await exchange.publish( + message=pika_message, routing_key=sanitize_as_rabbitmq_name(route.component_name) + ) logging_info( "###### RPC Request Published to exchg:(%s) routing_key=%s, msgId=%s", diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py index c2bce33..f8be7fc 100644 --- a/amqp/service/amq_service.py +++ b/amqp/service/amq_service.py @@ -4,7 +4,7 @@ import logging.config import os from asyncio import Future from threading import Thread -from typing import Optional +from typing import AsyncIterator, Optional from aio_pika.abc import AbstractRobustExchange @@ -12,7 +12,7 @@ from amqp.adapter.amq_route_factory import AMQRouteFactory from amqp.adapter.backpressure_handler import BackpressureHandler from amqp.adapter.consul_global_id_generator import get_unique_instance_id from amqp.adapter.data_message_factory import DataMessageFactory -from amqp.adapter.logging_utils import logging_info, logging_warning +from amqp.adapter.logging_utils import logging_error, logging_info, logging_warning from amqp.adapter.service_message_factory import ServiceMessageFactory from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import AMQRoute, DataMessage @@ -65,6 +65,7 @@ class AMQService: self.backpressure_thread: Optional[Thread] = None self.backpressure_handler: Optional[BackpressureHandler] = None self.maximum_availability = -1 + self.future: Future = asyncio.Future() # ======================================================================= # ======================================================================= @@ -82,6 +83,7 @@ class AMQService: logging.info("***********************************************") logging.info("****** AMQ Service Init Complete *********") logging.info("***********************************************") + self.future.set_result(True) self.loop.run_forever() def backpressure(self, current_availability: int, maximum: int = -1): @@ -163,6 +165,48 @@ class AMQService: return True return False + async def data_message_generator(self, route: AMQRoute) -> AsyncIterator[DataMessage]: + """ + An async iterator that blocks until messages arrive, processes them, + and continues in a loop. + + :param queue_name: The name of the queue to consume messages from + :yield: The processed DataResponse for each message + """ + # Get a channel from the connection + channel = self.rabbit_mq_client.channel + queue_args = ( + self.rabbit_mq_client.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None + ) + queue = await channel.declare_queue( + route.queue, + durable=True, + exclusive=False, + auto_delete=False, + arguments=queue_args, + ) + await queue.bind(route.exchange, route.key) + # Start consuming without a callback + async with queue.iterator() as queue_iter: + async for message in queue_iter: + try: + # Process the message using the same logic as in inbound_data_message_callback_no_thread + data_message: DataMessage = ( + await self.amq_data_message_handler.reconstitute_data_message(message) + ) + + success = yield data_message + if not success: + logging_warning(f"Message processing failed for {data_message.id()}.") + # Nack the message without requeuing in case of failure + await message.nack(requeue=False) + else: + await message.ack() + except Exception as e: + logging_error(f"Error processing message: {e}") + # Nack the message without requeuing in case of error + await message.nack(requeue=False) + def get_unique_instance_id(self) -> int: """ Provides cluster-wide unique instance ID for this AMQ service instance. diff --git a/demo.py b/demo.py index 25c48dc..7fb7cee 100644 --- a/demo.py +++ b/demo.py @@ -218,7 +218,12 @@ async def get_documents(): ) otadapter.amq_service.command( "otdemo-2", - "otdemo.otdemo_backend.__global__.print_document_id_static", + "print_document_id_static", + document_id=did + "-static", + ) + otadapter.amq_service.command( + "otdemo-2", + "print_document_id", document_id=did, ) diff --git a/otdemo/otdemo.backend.properties b/otdemo/otdemo.backend.properties new file mode 100644 index 0000000..71704b6 --- /dev/null +++ b/otdemo/otdemo.backend.properties @@ -0,0 +1,21 @@ +# ==================================================================== +# CleverMicro AMQ Adapter settings +# ==================================================================== +[CleverMicro-AMQ] +cm.amq-adapter.service-name=otdemo-2 +cm.amq-adapter.generator-id=1 +cm.amq-adapter.route-mapping=otdemo-2:otdemo-exchange:otdemo-queue-2:otdemo-2.#:5:0 +cm.amq-adapter.require-authenticated-user=false +cm.amq-adapter.callbacks-enabled=False + +# CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding +cm.dispatch.use-dlq=true +cm.dispatch.amq-host=localhost +cm.dispatch.amq-port=5672 +cm.dispatch.amq-port-tls=5671 +cm.dispatch.download-dir=/tmp/downloaded_files + +cm.backpressure.threshold=1 +cm.backpressure.time-window=150 +cm.backpressure.cpu-overload-duration=300 +cm.backpressure.cpu-idle-duration=300 diff --git a/otdemo/otdemo.properties b/otdemo/otdemo.properties index 4f4a8f5..a0a1695 100644 --- a/otdemo/otdemo.properties +++ b/otdemo/otdemo.properties @@ -2,21 +2,11 @@ # CleverMicro AMQ Adapter settings # ==================================================================== [CleverMicro-AMQ] -# 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=otdemo -# 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-dev route -cm.amq-adapter.route-mapping=otdemo:otdemo-exchange:otdemo-queue-1:api.cm.dev.cleverthis.localhost.test.python.frontend.#:5:0,otdemo-2:otdemo-exchange:otdemo-queue-2:otdemo-2.#:5:0 -# local route -#cm.amq-adapter.route-mapping=otdemo::otdemo:api.dev.localhost.otdemo.#:5:0 -# Indicate if AMQ Adapter should respond with HTTP 401 Unauthorized if the request does not contain valid JWT token +cm.amq-adapter.route-mapping=otdemo:otdemo-exchange:otdemo-queue-1:api.cm.dev.cleverthis.localhost.test.python.frontend.#:5:0 cm.amq-adapter.require-authenticated-user=false +cm.amq-adapter.callbacks-enabled=True # CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding cm.dispatch.use-dlq=true @@ -24,21 +14,8 @@ cm.dispatch.amq-host=localhost 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. -# 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-port=8080 -# Backpressure monitor settings -# Define maximum number of messages of being processed in parallel before triggering backpressure OVERLOAD alert cm.backpressure.threshold=1 -# 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=15 -# 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.time-window=150 +cm.backpressure.cpu-overload-duration=300 +cm.backpressure.cpu-idle-duration=300 diff --git a/otdemo/otdemo_backend.py b/otdemo/otdemo_backend.py index e9054dd..ee75a78 100644 --- a/otdemo/otdemo_backend.py +++ b/otdemo/otdemo_backend.py @@ -1,3 +1,10 @@ +import asyncio + +from amqp.config.amq_configuration import AMQConfiguration +from amqp.model.model import AMQRoute, DataMessage +from otdemo.otdemo_adapter import OTDemoAmqpAdapter + + def print_document_id_static(document_id: str) -> str: print(f"STATIC Document ID: {document_id}") return "STATIC PRINTED!" @@ -5,21 +12,42 @@ def print_document_id_static(document_id: str) -> str: class Backend: - def __init__(self, name: str, description: str): - self.name = name - self.description = description - - def __repr__(self): - return f"Backend(name={self.name}, description={self.description})" - - def __str__(self): - return f"{self.name}: {self.description}" - - def __eq__(self, other): - if not isinstance(other, Backend): - return False - return self.name == other.name and self.description == other.description - def print_document_id(self, document_id: str) -> str: print(f"CLASS Document ID: {document_id}") return "CLASS PRINTED!" + + +async def processing_loop(): + while ( + not otadapter.amq_service.future.done() + ): # Wait until AMQ adapter initializes RabbitMQ connection + await asyncio.sleep(0.1) + # IMPORTANT - in the next line the string arg must match the first token in route-mapping property + route: AMQRoute = otadapter.amq_service.router.find_route_by_service("otdemo-2") + generator = otadapter.amq_service.data_message_generator(route) # this is a generator function + print(f"Data message generator created: {generator}") + message = await generator.asend(None) # First value must be None to start the generator + while True: + print(f"Received message: {message}") + if message is None: + break + data_message: DataMessage = message + # This is a message processing part. You may use data_message.method() or .domain() or .path() + # to access values provided by caller in the send command() call. + document_id = data_message._base64body.decode("utf-8") + print( + f"DATA_MESSAGE Attributes: {data_message.method()} {data_message.domain()} {data_message.path()}" + ) + if data_message.path() == "print_document_id_static": + print_document_id_static(document_id) + elif data_message.path() == "print_document_id": + backend = Backend() + backend.print_document_id(document_id) + message = await generator.asend(True) + + +if __name__ == "__main__": + otadapter = OTDemoAmqpAdapter(AMQConfiguration("otdemo.backend.properties"), []) + # Run the processing loop in the AMQ service event loop + asyncio.run_coroutine_threadsafe(processing_loop(), otadapter.amq_service.loop) + otadapter.thread.join() # if the code above is not running a loop, need to join the thread to keep program running diff --git a/pyproject.toml b/pyproject.toml index d2693a8..f357526 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "amq_adapter" -version = "0.2.36" +version = "0.2.37" description = "The Python implementation of the AMQ Adapter for CleverMicro" readme = "README.md" authors = [