clevermicro/amq-adapter-python
14
0
Fork 0
Code Issues 16 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

rewrite using asynch PIKA library

Browse Source
This commit is contained in:
Stanislav Hejny
2025-03-13 02:13:56 +00:00
parent a7d7832a49
commit 5985a22bbb
11 changed files with 263 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amqp/adapter/cleverthis_service_adapter.py
+34 -6
View File
@@ -1,8 +1,11 @@
import asyncio
import json
import logging
import python_multipart
import io
import os
import re
import threading
from aiohttp import ClientSession, ClientResponse
from fastapi import HTTPException, UploadFile
@@ -16,6 +19,11 @@ from amqp.adapter.amq_response_factory import AMQResponseFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import AMQMessage, AMQResponse
# Track the number of messages currently being processed
current_parallel_executions = 0
lock = threading.Lock()
parallel_workers = os.getenv("PARALLEL_WORKERS", default=10)
class CleverMultiPartParser:
def __init__(self, message: AMQMessage):
@@ -183,6 +191,14 @@ class CleverThisServiceAdapter:
"""
return None
def sync_on_message(self, message: AMQMessage) -> AMQResponse:
"""
Synchronous version of on_message method. This is a wrapper around the async on_message method.
:param message: AMQMessage
:return: AMQResponse
"""
return asyncio.run(self.on_message(message))
async def on_message(self, message) -> AMQResponse:
"""
Called when a message is received from the AMQP.
@@ -190,6 +206,14 @@ class CleverThisServiceAdapter:
:return: AMQP response class with operation status code and optional error details.
"""
DEBUG_NO_AUTH = 1
global current_parallel_executions
# Increment the counter for parallel executions
with lock:
current_parallel_executions += 1
if current_parallel_executions > parallel_workers - 2:
logging.warn("Warning: CApacity close to depleted!")
_auth: str = message.headers.get('Authorization', '')
_auth_user: Any | None = None
if 'Bearer' in _auth:
@@ -198,22 +222,26 @@ class CleverThisServiceAdapter:
_auth_user = await self.get_current_user(token)
except Exception as error:
logging.error(f"Error getting user: {error}")
amq_response: AMQResponse | None = None
if _auth_user or DEBUG_NO_AUTH == 1:
method = message.method.lower()
if method == 'get':
return await self.get(message, _auth_user)
amq_response = await self.get(message, _auth_user)
elif method == 'put':
return await self.put(message, _auth_user)
amq_response = await self.put(message, _auth_user)
elif method == 'post':
return await self.post(message, _auth_user)
amq_response = await self.post(message, _auth_user)
elif method == 'head':
return await self.head(message, _auth_user)
amq_response = await self.head(message, _auth_user)
elif method == 'delete':
return await self.delete(message, _auth_user)
amq_response = await self.delete(message, _auth_user)
else:
raise ValueError(f"Unexpected HTTP method: {message.method}")
else:
return AMQResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info)
amq_response = AMQResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info)
with lock:
current_parallel_executions -= 1
return amq_response
# ==================================================================================================
# ===================== C l e v e r T h i s A P I m e t h o d s ============================
amqp/rabbitmq/rabbit_mq_consumer.py
+35 -94
View File
@@ -1,7 +1,8 @@
import logging
import pika
from pika.exchange_type import ExchangeType
from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange
from pika.exchange_type import ExchangeType
from amqp.model.model import AMQRoute
from amqp.rabbitmq.rabbit_mq_producer import RabbitMQProducer
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE, DLQ_EXCHANGE, AMQ_RPC_EXCHANGE
@@ -15,6 +16,20 @@ class RabbitMQConsumer(RabbitMQProducer):
def __init__(self, configuration):
super().__init__(configuration, RouterConsumer(configuration))
self.reply_to_exchange: AbstractRobustExchange | None = None # when sending response to RPC request
self.reply_to_queue: AbstractRobustQueue | None= None # when receiving response to RPC request
async def init(self, loop) -> AbstractRobustExchange:
await self.setup_amq_channel(loop)
cm_reply_to_queue = self.router.get_reply_to_queue_name()
self.reply_to_queue = await self.channel.declare_queue(
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=pika.exchange_type.ExchangeType.direct
)
await self.reply_to_queue.bind(exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue)
return self.reply_to_exchange
def cleanup(self):
logging.info(" [*] RabbitMQConsumer.cleanup()")
@@ -22,118 +37,44 @@ class RabbitMQConsumer(RabbitMQProducer):
if self.router.is_open(self.channel):
try:
self.channel.close()
except (pika.connection.exceptions.AMQPError, pika.connection.exceptions.ChannelClosedByBroker) as e:
except Exception as e:
logging.error(" [E] RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e)
def register_inbound_routes(self, callback):
async def register_inbound_routes(self, route_mapping: str, callback):
if self.router.is_open(self.channel):
queue_args = self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None
route_mapping = self.amq_configuration.amq_adapter.route_mapping
logging.info(" [*] RabbitMQConsumer register routes, cfg mapping: [%s]", route_mapping)
requested_routes = self.router.unwrap_route_list(route_mapping)
for route in requested_routes:
try:
queue_name = route.queue if route.queue else self.router.get_consume_queue_name()
logging.info(" [*] RabbitMQConsumer register route: [%s]", route)
self.channel.queue_declare(queue_name, durable=True, exclusive=False,
auto_delete=False, arguments=queue_args)
exchange = route.exchange if route.exchange else AMQ_RPC_EXCHANGE
self.channel.exchange_declare(exchange, exchange_type=pika.exchange_type.ExchangeType.topic)
self.channel.queue_bind(queue_name, exchange, route.key)
self.channel.basic_consume(
queue=queue_name,
on_message_callback=callback,
auto_ack=False
queue: AbstractRobustQueue = await self.channel.declare_queue(name=queue_name,
durable=True,
exclusive=False,
auto_delete=False,
arguments=queue_args)
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
)
self.router.register_route(
await queue.bind(exchange, route.key)
await queue.consume(callback, no_ack=False)
await self.router.register_route(
AMQRoute(
route.component_name, exchange, queue_name,
route.component_name, exchange_name, queue_name,
sanitize_as_rabbitmq_name(route.key),
route.timeout, route.valid_until
)
)
logging.info(" [*] AMQ connection initialized for route: %s, listens on: %s/%s",
route, exchange, queue_name)
except pika.connection.exceptions.AMQPError as err:
route, exchange, queue_name)
except Exception as err:
logging.error(" [E] Callback registration INCOMPLETE, %s", err)
else:
logging.warning(" [W] Channel is null, cannot register routes.")
def register_rpc_handler(self, rpc_consumer):
cm_reply_to_queue = self.router.get_reply_to_queue_name()
self.channel.queue_declare(cm_reply_to_queue, durable=True, exclusive=False, auto_delete=False)
self.channel.exchange_declare(
AMQ_REPLY_TO_EXCHANGE, exchange_type=pika.exchange_type.ExchangeType.direct
async def register_rpc_handler(self, rpc_consumer):
await self.reply_to_queue.consume(callback=rpc_consumer,
no_ack=False
)
self.channel.queue_bind(cm_reply_to_queue, AMQ_REPLY_TO_EXCHANGE, cm_reply_to_queue)
self.channel.basic_consume(
queue=cm_reply_to_queue,
on_message_callback=rpc_consumer,
auto_ack=False
)
"""
def limit_rate_inside_time_window(self, message: AMQMessage) -> int:
next_batch_size = BackpressureSubscriber.NO_OP
now = time.time_ns() // 1000000
queue_size = self.getQueueSize(message)
# 1. Backpressure? if unprocessed count exceeds threshold, wait for some time
if queue_size > THRESHOLD:
# TODO - signal queue backpressure explicitly to upstream/metrics
# wait: estimated-per-message-processing-time * SQRT(over-the-limit)
self.wait_for_milliseconds(
(WINDOW / self.request_for_window_duration) * (queue_size - THRESHOLD) ** 0.5)
queue_size = self.getQueueSize(message) # refresh queue size after wait
# 2. batch complete? determine next batch size and wait till window end. reset window.
if self.fulfilled >= self.request_for_window_duration:
next_batch_size = self.calculateNextBatchSize(queue_size)
self.request_for_window_duration = next_batch_size
remaining_time = (self.window_started_at + WINDOW) - now
if remaining_time > 0:
self.wait_for_milliseconds(remaining_time)
queue_size = self.getQueueSize(message) # refresh queue size after wait
self.window_started_at = time.time_ns() // 1000000
self.fulfilled = 0
# 3. Window elapsed? Reset window & recalculate batch size
if now >= self.window_started_at + WINDOW:
if next_batch_size == BackpressureSubscriber.NO_OP:
next_batch_size = self.calculateNextBatchSize(queue_size)
self.request_for_window_duration = next_batch_size
self.window_started_at = time.time_ns() // 1000000
self.fulfilled = 0
# 4. send the message from the current batch
self.send_message(message)
self.fulfilled += 1
# 5. report and setup state for the next call
end = time.time_ns() // 1000000
self.on_next_timestamp = end
return next_batch_size
def calculateNextBatchSize(self, queue_size: int) -> int:
next_batch_size = self.request_for_window_duration
if queue_size <= THRESHOLD:
# processing proceeds well, can increase the rate
next_batch_size += self.subscriber.increaseRate(self.request_for_window_duration)
logging.debug(f"INCREASE rate to {next_batch_size}. qSize={queue_size}")
else:
decrease = self.subscriber.backoff(queue_size - THRESHOLD)
if self.request_for_window_duration > decrease:
next_batch_size -= decrease
logging.debug(f"DECREASE rate to {next_batch_size}. qSize={queue_size}")
return next_batch_size
"""
"""
def getQueueSize(self, message: AMQMessage) -> int:
try:
route_queue_name = self.router.findRoute(message)
queue_name = route_queue_name.queue if route_queue_name else self.TASK_DISPATCH_QUEUE_NAME
logging.info(f"Get Queue size for {queue_name}")
return max(len(self.awaitingACK), self.channel.message_count(queue_name))
except Exception:
return len(self.awaitingACK)
"""
amqp/rabbitmq/rabbit_mq_producer.py
+43 -85
View File
@@ -1,10 +1,5 @@
import logging
import time
import pika
from pika.adapters.blocking_connection import BlockingChannel, BlockingConnection
from pika.channel import Channel
from pika.connection import Connection
import aio_pika
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
@@ -14,38 +9,22 @@ from amqp.router.router_producer import RouterProducer
class RabbitMQProducer:
PREFETCH_COUNT = 1
THRESHOLD = 1000
WINDOW = 5000 # 5 seconds
def __init__(self, amq_configuration: AMQConfiguration, router: RouterProducer = None):
self.channel: BlockingChannel | None = None
self.connection: BlockingConnection | None = None
self.request_for_window_duration = None
self.channel: aio_pika.RobustChannel | None = None
self.connection: aio_pika.RobustConnection | None = None
self.rpc_exchange: aio_pika.RobustExchange | None = None
self.reply_exchange: aio_pika.RobustExchange | None = None
self.amq_configuration = amq_configuration
logging.info("*******************************************")
logging.info("******** RabbitMQProducer.init() ********** %s - %s - %s",
self.amq_configuration.dispatch.amq_host,
self.amq_configuration.dispatch.use_confirms,
self.WINDOW)
logging.info("******** RabbitMQProducer.init(): %s",
self.amq_configuration.dispatch.amq_host)
logging.info("*******************************************")
self.router = router or RouterProducer(amq_configuration)
# this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
self.setup_amq_channel()
self.router.init_routing_paths(self.channel)
# self.backpressure_per_exchange = defaultdict(lambda: Gauge(
# f"amqp.adapter.exchange.{exchange}", registry=CollectorRegistry()))
self.router.set_route_added_notifier(
lambda route: logging.info(" [DBG] ---------- RouteAdded, setting up metrics: amqp.adapter.exchange.%s",
route.exchange))
self.request_for_window_duration = self.THRESHOLD // 2
self.window_started_at = time.time()
self.on_next_timestamp = self.window_started_at
self.fulfilled = 0
async def init(self, loop):
await self.setup_amq_channel(loop)
def cleanup(self):
if self.connection:
@@ -54,30 +33,27 @@ class RabbitMQProducer:
except Exception as e:
logging.error("RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e)
def setup_amq_channel(self):
async def setup_amq_channel(self, loop):
"""
Set up the RabbitMQ connection
:return: nothing, it applies changes to instance variables
"""
credentials: pika.credentials.PlainCredentials = pika.credentials.PlainCredentials(
username=self.amq_configuration.dispatch.rabbit_mq_user,
password=self.amq_configuration.dispatch.rabbit_mq_password
)
factory = pika.ConnectionParameters(
self.connection = await aio_pika.connect_robust(
host=self.amq_configuration.dispatch.amq_host,
credentials=credentials
port=self.amq_configuration.dispatch.amq_port,
login=self.amq_configuration.dispatch.rabbit_mq_user,
password=self.amq_configuration.dispatch.rabbit_mq_password,
loop=loop
)
self.connection = pika.BlockingConnection(factory)
logging.info("RabbitMQProducer.setupAMQ, got connection, open=%s", self.connection.is_open)
self.connection.add_on_connection_blocked_callback(self.blocked_listener)
self.connection.add_on_connection_unblocked_callback(self.unblocked_listener)
self.channel = self.connection.channel()
self.channel.basic_qos(prefetch_count=self.PREFETCH_COUNT)
if self.amq_configuration.dispatch.use_confirms:
self.setup_local_ack()
self.channel.add_on_return_callback(self.return_listener)
logging.info("RabbitMQProducer.setupAMQ, got connection, open=%s", self.connection.is_closed)
self.channel = await self.connection.channel()
await self.channel.set_qos(prefetch_count=self.PREFETCH_COUNT)
await self.router.init_routing_paths(self.channel)
self.router.set_route_added_notifier(
lambda route: logging.info(" [DBG] ---------- RouteAdded, setting up metrics: amqp.adapter.exchange.%s",
route.exchange))
def send_message(self, message: AMQMessage):
async def send_message(self, message: AMQMessage):
"""
The API routine that sends a datagram message (AMQMessage) to the destination. The Destination
is determined from the AMQMessage metadata section, using `domain` and `path` as key.
@@ -87,27 +63,31 @@ class RabbitMQProducer:
"""
try:
route = self.router.find_route_by_message(message)
if route:
if route.timeout < 0:
if route and self.rpc_exchange:
if route.timeout <= 0:
# No response expected
self.channel.basic_publish(
route.exchange,
self.router.get_routing_key(message), # context path is a routing key
properties=pika.BasicProperties(delivery_mode=2),
body=AMQMessageFactory.serialize(message)
pika_message: aio_pika.message.Message = aio_pika.message.Message(
body=AMQMessageFactory.serialize(message),
content_encoding='utf-8',
delivery_mode=2
)
await self.rpc_exchange.publish(
message=pika_message,
routing_key=self.router.get_routing_key(message), # context path is a routing key
)
logging.info(f" [x] basicPublish (no-reply), messageId: {message.id}, to: {route}")
else:
# This is RPC call, there should be response
props = pika.BasicProperties(
correlation_id=message.id,
pika_message: aio_pika.message.Message = aio_pika.message.Message(
body=AMQMessageFactory.serialize(message),
content_encoding='utf-8',
delivery_mode=2,
correlation_id=str(message.id),
reply_to=self.router.get_reply_to_queue_name()
)
self.channel.basic_publish(
route.exchange,
self.router.get_routing_key(message), # context path is a routing key
properties=props,
body=AMQMessageFactory.serialize(message)
await self.rpc_exchange.publish(
message=pika_message,
routing_key=self.router.get_routing_key(message), # context path is a routing key
)
logging.info(" [x] basicPublish (RPC call), messageId: %s, to: %s",
message.id.as_string(), route.as_string())
@@ -117,30 +97,8 @@ class RabbitMQProducer:
except Exception as err:
logging.error("Send message failed: %s", err)
def request_routes(self):
self.router.request_routes_from_adapters()
def setup_local_ack(self):
self.channel.confirm_delivery()
async def request_routes(self):
await self.router.request_routes_from_adapters()
def find_route(self, domain: str, path: str) -> AMQRoute:
return self.router.find_route(domain, path)
def return_listener(self, return_message):
logging.error(
f" [E] ****** (NOT DELIVERED) ID={return_message.properties.message_id}, "
f"Exchg=[{return_message.exchange}], Key=[{return_message.routing_key}], "
f"replyTxt={return_message.reply_text}, length={len(return_message.body) if return_message.body else 0}"
)
def blocked_listener(self, reason: str):
logging.error(f" [E] CHANNEL BLOCKED, reason={reason}")
def unblocked_listener(self):
logging.info(" [*] CHANNEL UNBLOCKED")
def ack_nack_callback(method):
if method == pika.spec.Basic.Ack:
pass
if method == pika.spec.Basic.Nack:
pass
amqp/router/router_consumer.py
+21 -29
View File
@@ -1,9 +1,7 @@
import logging
from typing import Set
from pika import spec
from pika.channel import Channel
from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractIncomingMessage
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import CleverMicroServiceMessage, AMQRoute
@@ -30,25 +28,23 @@ class RouterConsumer(RouterProducer):
self.amqRouteFactory = AMQRouteFactory()
logging.info(" [*] RouterConsumer.<init>")
def init_routing_paths_consumer(self, _channel: Channel):
async def init_routing_paths_consumer(self, _channel: AbstractRobustChannel):
"""
* Initialize router according to Consumer mode.
"""
self.init_routing_paths_consumer(_channel)
await self.init_routing_paths_consumer(_channel)
logging.info(" [*] RouterConsumer.initConsumerRouter, channel open: %s", self.is_open(_channel))
if self.is_open(_channel):
try:
# 1. declare a Queue that gives the RoutingDAta requests
cm_request_queue: str = self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_REQ_QUEUE
self.channel.queue_declare(cm_request_queue, durable=True, exclusive=False, auto_delete=False)
queue: AbstractRobustQueue = await self.channel.declare_queue(name=cm_request_queue, durable=True, exclusive=False, auto_delete=False)
# 1a. bind queue to CM_RESPONSE_EXCHANGE w/o any conditions (empty routing key)
self.channel.queue_bind(queue=cm_request_queue, exchange=CM_REQUEST_EXCHANGE, routing_key="")
await queue.bind(exchange=CM_REQUEST_EXCHANGE, routing_key="")
# 1b. Service adapter mode - listens on Request queue and replies to Response queue
logging.info(" [*] RouterConsumer listens for RouteDataRequests on: %s", cm_request_queue)
self.channel.basic_consume(queue=cm_request_queue,
auto_ack=False,
on_message_callback=self.handle_routing_data_request_message)
await queue.consume(callback=self.handle_routing_data_request_message, no_ack=False)
except Exception as ioe:
logging.error(" [E] RouterConsumer FATAL: cannot initialize, %s", ioe)
else:
@@ -58,39 +54,35 @@ class RouterConsumer(RouterProducer):
* ServiceMessageType.ROUTING_DATA_REQUEST handler listening on queue 'CleverMicroRequest'
"""
def handle_routing_data_request_message(self,
cb_channel: Channel,
method: spec.Basic.Deliver,
properties: spec.BasicProperties,
body: bytes):
async def handle_routing_data_request_message(self, message: AbstractIncomingMessage):
# some debug statements, TODO - remove for prod release
delivery_tag = method.delivery_tag
debug = "Delivery.Properties = " + properties.__str__()
delivery_tag = message.delivery_tag
debug = "Delivery.Properties = " + message.properties.__str__()
logging.info(" [DBG] ******[%] (ROUTING_DATA_REQ) tag: %s, env: %s, props: %s",
delivery_tag, method.consumer_tag, method.get_properties(), debug)
delivery_tag, message.consumer_tag, message.properties, debug)
cb_channel.basic_ack(delivery_tag, False)
await message.ack(False)
message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(body)
cm_message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(message.body)
route_mapping: str = self.wrap_own_routes()
logging.info(" [DBG] ******[%s] (ROUTING_DATA_REQ) msg=%s, routeMap=[%s]",
delivery_tag, self.service_message_factory.to_string(message), route_mapping)
delivery_tag, self.service_message_factory.to_string(cm_message), route_mapping)
# Reply only if self adapter has own routeMapping (e.g. API-gateway-side adapter has none)
if message.is_valid() and message.message_type == ServiceMessageType.ROUTING_DATA_REQUEST and route_mapping:
if cm_message.is_valid() and cm_message.message_type == ServiceMessageType.ROUTING_DATA_REQUEST and route_mapping:
try:
service_message: CleverMicroServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA, route_mapping, message.reply_to
ServiceMessageType.ROUTING_DATA, route_mapping, cm_message.reply_to
)
self.publish_service_message(service_message, CM_RESPONSE_EXCHANGE)
await self.publish_service_message(service_message, self.cm_response_exchange)
except Exception as err:
logging.error(" [E] ******[%s] (ROUTING_DATA_REQ) Response to queue: %s failed: %s",
delivery_tag, CM_RESPONSE_EXCHANGE, err)
else:
logging.error(" [E] ******[%s] (ROUTING_DATA_REQ) [%s] produces invalid or no CM msg type.",
delivery_tag, str(body))
delivery_tag, str(message.body))
def register_route(self, route: AMQRoute):
async def register_route(self, route: AMQRoute):
"""
* Called in Consumer (Service Adapter) context. Registers the routing data for self Service
* with the Master router by sending own routing data to Master via CleverMicroResponse queue.
@@ -101,7 +93,7 @@ class RouterConsumer(RouterProducer):
serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA, route.as_string(), ANY_RECIPIENT
)
self.publish_service_message(serviceMessage, CM_RESPONSE_EXCHANGE)
await self.publish_service_message(serviceMessage, self.cm_response_exchange)
self.add_own_route(route)
logging.info(" [DBG] RouterConsumer registered Route: %s", route)
else:
@@ -110,7 +102,7 @@ class RouterConsumer(RouterProducer):
except Exception as ioe:
logging.error(" [E] RouterConsumer failed register route data: %s", ioe)
def remove_route_on_cleanup(self, route: AMQRoute) -> bool:
async def remove_route_on_cleanup(self, route: AMQRoute) -> bool:
"""
* Unregister / remove route from the list
* :param route: route to remove
@@ -123,7 +115,7 @@ class RouterConsumer(RouterProducer):
route.as_string(),
ANY_RECIPIENT
)
if not self.publish_service_message(service_message, CM_RESPONSE_EXCHANGE):
if not await self.publish_service_message(service_message, self.cm_response_exchange):
logging.warning(" [W] RouterConsumer couldn't remove current route: %s, channel not open.",
route)
amqp/router/router_producer.py
+52 -61
View File
@@ -7,10 +7,10 @@
* forwards it to the application.
"""
import logging
import os
import pika
from pika import BasicProperties, spec
from pika.channel import Channel
from aio_pika import Message
from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractRobustConnection, AbstractRobustExchange, \
AbstractIncomingMessage
from pika.exchange_type import ExchangeType
from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory
@@ -29,26 +29,18 @@ class RouterProducer(RouterBase):
* :param configuration: adapter's configuration reference.
"""
super().__init__()
self.connection = None
self.connection: AbstractRobustConnection | None = None
self.service_message_factory = CleverMicroServiceMessageFactory(configuration.amq_adapter.service_name)
self.amq_configuration: AMQConfiguration = configuration
self.channel: Channel | None = None
self.channel: AbstractRobustChannel | None = None
self.cm_request_exchange: AbstractRobustExchange | None = None
self.cm_response_exchange: AbstractRobustExchange | None = None
logging.info(f" [*] RouterProducer.<init>: %s, route: %s",
self.amq_configuration.amq_adapter.service_name,
self.amq_configuration.amq_adapter.route_mapping)
def amq_consumer(self):
credentials = pika.PlainCredentials(
os.environ['RABBIT_MQ_USER'], os.environ['RABBIT_MQ_PASSWORD']
)
self.connection: pika.SelectConnection = pika.SelectConnection(
pika.ConnectionParameters(host=self.amq_configuration.dispatch.amq_host, credentials=credentials)
)
self.channel = self.connection.channel()
# self.channel.basic_consume()
def init_routing_paths(self, channel: Channel | None):
async def init_routing_paths(self, channel: AbstractRobustChannel | None):
"""
* Initialize router according to Producer mode.
* The logic here sets a producer and listener on CleverMicroRequest/CleverMicroResponse queues
@@ -63,22 +55,25 @@ class RouterProducer(RouterBase):
try:
# **** set up the service discovery internal exchanges ****
# 1. declare RequestExchange to where to publish the RoutingData requests
self.channel.exchange_declare(exchange=CM_REQUEST_EXCHANGE, exchange_type=ExchangeType.fanout)
self.cm_request_exchange = await self.channel.declare_exchange(
name=CM_REQUEST_EXCHANGE, type=ExchangeType.fanout
)
#
# 2. declare ResponseExchange where to publish current RoutingData (response to
# a RoutingData request)
self.channel.exchange_declare(exchange=CM_RESPONSE_EXCHANGE, exchange_type=ExchangeType.fanout)
self.cm_response_exchange = await self.channel.declare_exchange(
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()
self.channel.queue_declare(cm_response_queue, durable=True, exclusive=False, auto_delete=False,
arguments={})
response_queue: AbstractRobustQueue = await self.channel.declare_queue(
name=cm_response_queue, durable=True, exclusive=False, auto_delete=False,
arguments={}
)
# 2b. bind queue to CM_RESPONSE_EXCHANGE w/o any conditions (empty routing key)
self.channel.queue_bind(queue=cm_response_queue, exchange=CM_RESPONSE_EXCHANGE, routing_key="#")
await response_queue.bind(exchange=CM_RESPONSE_EXCHANGE, routing_key="#")
# 2c. listen for incoming RoutingData messages
self.channel.basic_consume(
queue=cm_response_queue,
auto_ack=False,
on_message_callback=self.handle_routing_data_message)
await response_queue.consume(callback=self.handle_routing_data_message, no_ack=False)
logging.info(" [*] Routing master listens for Route updates on %s", cm_response_queue)
except Exception as ioe:
logging.error(" [E] RouterProducer FATAL: cannot initialize, %s", ioe)
@@ -86,38 +81,34 @@ class RouterProducer(RouterBase):
else:
logging.error(" [E] RouterProducer FATAL: cannot initialize, RabbitMQ channel is not open.")
def handle_routing_data_message(self,
cb_channel: Channel,
method: spec.Basic.Deliver,
properties: spec.BasicProperties,
body: bytes):
async def handle_routing_data_message(self, message: AbstractIncomingMessage):
"""
* Handler for CLEVER_MICRO_RESPONSE queue messages. Used in Master mode to handle receipt of
* routing data, invokes addRoute or removeRoute depending on the type of the message.
"""
logging.info(f" [x] Received {body}, m={method}, p={properties}")
logging.info(f" [x] Received {message.body}, m={message.message_id}, p={message.properties}")
"""
logging.info(" [DBG] ******[%s] ROUTING_DATA %s >>> %s",
delivery.getEnvelope().getDeliveryTag(),
delivery.getEnvelope().toString(), str(delivery.getBody()))
"""
# ACK message now so that it is not being redelivered
cb_channel.basic_ack(method.delivery_tag, False)
message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(body)
await message.ack(multiple=False)
cm_message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(message.body)
# ignore the message if it comes from ourselves (even if from different instance)
logging.info(" [DBG] ******[%s] ROUTING_DATA, msg=%s, routeMap=[%s] sn=%s, replyTo=%s",
method.delivery_tag,
self.service_message_factory.to_string(message),
message.delivery_tag,
self.service_message_factory.to_string(cm_message),
self.amq_configuration.amq_adapter.route_mapping,
self.amq_configuration.amq_adapter.service_name,
message.reply_to)
if message.is_valid() and not self.amq_configuration.amq_adapter.service_name == message.reply_to:
if message.message_type == ServiceMessageType.ROUTING_DATA:
self.add_routes(message.message)
if message.message_type == ServiceMessageType.ROUTING_DATA_REMOVE:
self.remove_routes(message.message)
cm_message.reply_to)
if 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)
if cm_message.message_type == ServiceMessageType.ROUTING_DATA_REMOVE:
self.remove_routes(cm_message.message)
def request_routes_from_adapters(self):
async def request_routes_from_adapters(self):
"""
* Send a broadcast message prompting all listening adapters to reply with own routing data.
"""
@@ -126,37 +117,37 @@ class RouterProducer(RouterBase):
serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT
)
if not self.publish_service_message(serviceMessage, CM_REQUEST_EXCHANGE):
if not await self.publish_service_message(serviceMessage, self.cm_request_exchange):
logging.warning(" [E] RouterProducer could not request route data: %s channel is not open.",
CM_REQUEST_EXCHANGE)
except Exception as ioe:
logging.error(" [E] RouterProducer failed request routing data: %s", ioe)
BASIC: BasicProperties = BasicProperties(content_type="application/octet-stream",
content_encoding=None,
headers=None,
delivery_mode=1,
priority=0,
correlation_id=None)
def publish_service_message(self, message: CleverMicroServiceMessage, exchange_name: str) -> bool:
async def publish_service_message(self,
message: CleverMicroServiceMessage,
exchange: AbstractRobustExchange) -> bool:
"""
* Publishes a service message to the service queue and logs success log entry.
*
* :param message: Service Message to be sent
* :param exchangeName: target exchange name where to publish the message
* :param queueName target: queue name (Optional, purpose is to get the queue size only)
* :param exchange: target exchange name where to publish the message
* :return True if channel is open, False otherwise
* @throws IOException if something else goes wrong
"""
if self.is_open(self.channel):
binary_content: bytes = self.service_message_factory.serialize(message)
self.channel.basic_publish(exchange=exchange_name,
routing_key="", # empty routing key
properties=self.BASIC,
body=binary_content)
logging.info(" [x] Service Message Sent to %s, msg: %s", exchange_name, str(binary_content))
pika_message: Message = Message(
body=binary_content,
content_encoding='utf-8',
delivery_mode=2,
content_type="application/octet-stream",
headers=None,
priority=0,
correlation_id=None
)
await exchange.publish(message=pika_message, routing_key="")
logging.info(" [x] Service Message Sent to %s, msg: %s", exchange.name, str(binary_content))
return True
return False
@@ -181,10 +172,10 @@ class RouterProducer(RouterBase):
"""
return self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_AMQ_QUEUE
def is_open(self, _channel: Channel) -> bool:
def is_open(self, _channel: AbstractRobustChannel) -> bool:
"""
* Test if the channel is usable (opened).
* :param _channel: the channel
* :return True if the channel is opened.
"""
return _channel is not None and _channel.is_open
return _channel is not None and not _channel.is_closed
amqp/service/amq_service.py
+27 -12
View File
@@ -1,8 +1,10 @@
import logging
import logging.config
import os
import asyncio
from asyncio import Future
from aio_pika.abc import AbstractRobustExchange
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.amq_route_factory import AMQRouteFactory
@@ -29,53 +31,66 @@ 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, http_adapter):
def __init__(self, amq_configuration: AMQConfiguration, service_adapter):
logging.info("***********************************************")
logging.info("********** AMQ Service *******************")
logging.info("***********************************************")
loop = asyncio.get_event_loop()
self.amq_configuration = amq_configuration
self.http_adapter = http_adapter
self.message_factory = AMQMessageFactory.get_instance(self.amq_configuration.amq_adapter.generator_id)
self.rabbit_mq_consumer = RabbitMQConsumer(self.amq_configuration)
self.tracer = initialize_trace(name=amq_configuration.amq_adapter.service_name)
self.amq_service_message_handler = AMQServiceMessageHandler(self.http_adapter, self.tracer, self.message_factory)
self.amq_service_message_handler = None
self.service_message_factory = CleverMicroServiceMessageFactory(
self.amq_configuration.amq_adapter.service_name
)
self.register_routes()
# self.init(loop) will initialize RabbitMQ connection
loop.run_until_complete(self.init(loop, service_adapter))
# loop.run_until_complete(consume(loop))
# self.consumer_thread = asyncio.create_task(self.rabbit_mq_consumer.channel.start_consuming())
# self.rabbit_mq_consumer.request_routes()
logging.info("***********************************************")
logging.info("****** AMQ Service Init Complete *********")
logging.info("***********************************************")
loop.run_forever()
async def init(self, loop, service_adapter):
exchange: AbstractRobustExchange = await self.rabbit_mq_consumer.init(loop)
self.amq_service_message_handler = AMQServiceMessageHandler(
service_adapter, self.tracer, self.message_factory, exchange
)
await self.register_routes()
await self.rabbit_mq_consumer.request_routes()
def cleanup(self):
logging.info(" [DBG] AMQService.cleanup() rmqc=%s", self.rabbit_mq_consumer)
if self.rabbit_mq_consumer:
self.rabbit_mq_consumer.cleanup()
def reinitialize_if_disconnected(self):
if not self.rabbit_mq_consumer.channel or not self.rabbit_mq_consumer.channel.is_open:
async def reinitialize_if_disconnected(self):
if not self.rabbit_mq_consumer.channel or self.rabbit_mq_consumer.channel.is_closed:
self.rabbit_mq_consumer = RabbitMQConsumer(self.amq_configuration)
self.register_routes()
self.rabbit_mq_consumer.request_routes()
await self.rabbit_mq_consumer.init(loop=None)
await self.register_routes()
await self.rabbit_mq_consumer.request_routes()
def register_routes(self):
async def register_routes(self) -> AbstractRobustExchange | None:
route_mapping = self.amq_configuration.amq_adapter.route_mapping
if AMQRouteFactory.validate(route_mapping):
try:
self.rabbit_mq_consumer.register_inbound_routes(
self.amq_service_message_handler.inbound_message_callback_http
await self.rabbit_mq_consumer.register_inbound_routes(
route_mapping, self.amq_service_message_handler.inbound_message_callback
)
self.rabbit_mq_consumer.register_rpc_handler(
return await self.rabbit_mq_consumer.register_rpc_handler(
self.amq_service_message_handler.reply_received_callback
)
except Exception as e:
raise RuntimeError(e)
elif route_mapping:
logging.info(" [W] route [%s] failed validation, no routes were registered.", route_mapping)
return None
def send_message(self, message: AMQMessage, future: Future):
self.amq_service_message_handler.outstanding[str(message.id)] = future
amqp/service/service_message_handler.py
+48 -53
View File
@@ -1,23 +1,28 @@
import asyncio
import logging
import time
import concurrent.futures
import os
from asyncio import Future
from typing import Dict
import pika
from aio_pika import IncomingMessage, Message
from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange
from opentelemetry import trace
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.trace import Tracer, TraceState
from pika.channel import Channel
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.amq_response_factory import AMQResponseFactory
from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter
from amqp.model.model import AMQResponse, AMQMessage
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
from amqp.router.serializer import map_as_string
parallel_workers = os.getenv("PARALLEL_WORKERS", default=10)
# Shared thread pool for parallel execution
executor = concurrent.futures.ThreadPoolExecutor(max_workers=parallel_workers)
def collect_trace_info():
trace_map = {}
@@ -37,37 +42,32 @@ def set_text(carrier, key, value):
class AMQServiceMessageHandler:
def __init__(self, service_adapter: CleverThisServiceAdapter, tracer: Tracer, message_factory: AMQMessageFactory):
self.http_adapter: CleverThisServiceAdapter = service_adapter
def __init__(self, service_adapter: CleverThisServiceAdapter,
tracer: Tracer, message_factory: AMQMessageFactory,
reply_to_exchange: AbstractRobustExchange):
self.service_adapter: CleverThisServiceAdapter = service_adapter
self.tracer = tracer
self.message_factory: AMQMessageFactory = message_factory
self.reply_to_exchange: AbstractRobustExchange = reply_to_exchange
# Dictionary to store outstanding Futures
self.outstanding: Dict[str, asyncio.Future] = {}
def inbound_message_callback_http(self,
channel: Channel,
method: pika.spec.Basic.Deliver,
properties: pika.spec.BasicProperties,
body: bytes):
async def inbound_message_callback(self, message: AbstractIncomingMessage):
"""
Receives the message as bytes from RabbitMQ queue and deserializes it to the AMQMessage
Ensures Jaeger trace-id propagation.
In loose coupling mode, invokes appropriate adapter (in v0.1 only HTTPAdapter is supported).
In tight coupling mode, invokes custom handler/mapper/adapter method that
:param body:
:param properties:
:param method:
:param channel:
Invokes custom handler/mapper/adapter method that corresponds to this message
:param message:
:return:
"""
start = time.time()
delivery_tag = method.delivery_tag
debug = f" [DBG] ######[{delivery_tag}] (AMQ MESSAGE), env: {method}, props: {properties}"
delivery_tag = message.delivery_tag
debug = f" [DBG] ######[{delivery_tag}] (AMQ MESSAGE), env: {message.headers}, props: {message.properties}"
logging.info(debug)
amq_message: AMQMessage = AMQMessageFactory.from_bytes(body)
amq_message: AMQMessage = AMQMessageFactory.from_bytes(message.body)
logging.info(" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, traceInfo=%s",
delivery_tag, method.consumer_tag, amq_message.id, map_as_string(amq_message.trace_info))
delivery_tag, message.consumer_tag, amq_message.id, map_as_string(amq_message.trace_info))
propagator = TraceContextTextMapPropagator()
context = propagator.extract(carrier=amq_message.trace_info, context=trace.context_api.get_current())
@@ -82,21 +82,25 @@ class AMQServiceMessageHandler:
# map_as_string(amq_message.trace_info), context_with_span, context)
if amq_message.magic == AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST:
response: AMQResponse = asyncio.run(self.http_adapter.on_message(amq_message))
self.send_reply(properties.reply_to, delivery_tag, channel, response)
future = executor.submit(self.service_adapter.sync_on_message, amq_message)
# response: AMQResponse = asyncio.run(self.http_adapter.on_message(amq_message))
# Wait for the result and send it back to the reply queue
try:
response: AMQResponse = future.result() # Block until the result is available
await self.send_reply(message.reply_to, delivery_tag, response)
except Exception as e:
print(f"[E] Error processing message: {e}")
await message.nack(requeue=False)
return
else:
logging.error(" [E] Unknown MAGIC value: [%s], don't know how to interpret the message",
amq_message.magic)
await message.nack(requeue=False)
return
duration = time.time() - start
logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, duration)
channel.basic_ack(delivery_tag, False)
# if "___NACK___" in str(amq_message.body()):
# logging.info(" [x] ######[%s] Done, single NACK: %sms.", delivery_tag, duration)
# channel.basic_nack(delivery_tag, False, False)
# else:
# logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, duration)
# channel.basic_ack(delivery_tag, False)
await message.ack(multiple=False)
def on_http_response(self, delivery, deliveryTag, channel, http_response):
"""
@@ -119,54 +123,45 @@ class AMQServiceMessageHandler:
collect_trace_info()
)
try:
self.send_reply(delivery.properties.reply_to, deliveryTag, channel, response)
self.send_reply(delivery.properties.reply_to, deliveryTag, response)
except Exception as e:
raise RuntimeError(e)
def send_reply(self, reply_to: str, delivery_tag: str, channel: Channel, response: AMQResponse):
async def send_reply(self, reply_to: str, delivery_tag: int | None, response: AMQResponse):
"""
The Service side code. When the CleverXXX service has output ready, in AMQResponse object,
call this method to return the reply back to the caller
:param reply_to:
:param delivery_tag:
:param channel:
:param response:
:return:
"""
if reply_to:
logging.info(" [*] ######[%s] AMQ Message Reply, To=%s, msgId=%s",
delivery_tag, reply_to, response.id)
reply_props = pika.BasicProperties(
pika_message: Message = Message(
body=AMQResponseFactory.serialize(response),
correlation_id=response.id,
content_type=response.content_type[0]
if isinstance(response.content_type, list) else response.content_type
)
channel.basic_publish(
exchange=AMQ_REPLY_TO_EXCHANGE,
routing_key=reply_to,
body=AMQResponseFactory.serialize(response),
properties=reply_props,
await self.reply_to_exchange.publish(
message=pika_message,
routing_key=reply_to
)
def reply_received_callback(self,
channel: Channel,
method: pika.spec.Basic.Deliver,
properties: pika.spec.BasicProperties,
body: bytes):
async def reply_received_callback(self, message: IncomingMessage):
"""
The producer/client (API Gateway) facing response_message recipient - must match the received response with
the original request and produce the HTTP reply.
Note: this code runs on the API Gateway side of RabbitMQ
:param body:
:param properties:
:param method:
:param channel:
:param message:
:return:
"""
reply_id = properties.correlation_id
content_type = properties.content_type
delivery_tag = method.delivery_tag
debug = f" [DBG] ######[{delivery_tag}] (RPC REPLY) env: {method}, props: {properties}, BODY: {body.decode()}"
reply_id = message.correlation_id
content_type = message.content_type
delivery_tag = message.delivery_tag
debug = f" [DBG] ######[{delivery_tag}] (RPC REPLY) env: {message.headers}, props: {message.properties}, BODY: {message.body.decode()}"
logging.info(debug)
# TODO - figure out implementation of this part, as this is sending reply back either to a service,
# or to the user
@@ -184,8 +179,8 @@ class AMQServiceMessageHandler:
# internal, service-2-service communication, and the proper response type will be decided than
# reply.response().set_result(Response.ok(result.body, content_type).build())
if future:
response: AMQResponse = AMQResponseFactory.from_bytes(body)
response: AMQResponse = AMQResponseFactory.from_bytes(message.body)
# Complete the Future with the response data
future.set_result(response.body())
channel.basic_ack(delivery_tag, False)
await message.ack(multiple=False)
amqp/service/tracing.py
-1
View File
@@ -1,5 +1,4 @@
from opentelemetry import metrics, trace
#from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.trace import TracerProvider, Tracer
from opentelemetry.sdk.trace.export import BatchSpanProcessor
pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "amq_adapter"
version = "0.2.4"
version = "0.2.5"
description = "The Python implementation of the AMQ Adaptor for CleverMicro"
readme = "README.md"
license = { text = "Apache License 2.0" }
requirements.txt
+1 -1
View File
@@ -1,6 +1,6 @@
aiohttp~=3.11.11
fastapi==0.115.6
pika~=1.3.2
aio_pika~=9.5.5
opentelemetry-api
opentelemetry-sdk
opentelemetry-exporter-otlp
setup.py
+1 -1
View File
@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
setup(
name="amqp",
version="0.2.4",
version="0.2.5",
author="Stanislav Hejny",
author_email="stanislav.hejny@cleverthis.com",
description="Adapter for RabbitMQ to integrate a CleverThis python-code Service with CleverMicro platform",