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

feat/#1 - applied 'black' formatting

Browse Source
This commit is contained in:
Stanislav Hejny
2025-04-18 23:49:10 +01:00
parent 04977e5ccc
commit 1de7afe4f5
35 changed files with 1239 additions and 563 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amqp/service/amq_message_handler.py
+142 -65
View File
@@ -23,7 +23,13 @@ from amqp.adapter.file_downloader import on_message
from amqp.adapter.logging_utils import logging_error
from amqp.adapter.service_message_factory import ServiceMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import DataResponse, DataMessage, MultipartDataMessage, ScalingRequest, ScalingRequestAlert
from amqp.model.model import (
DataResponse,
DataMessage,
MultipartDataMessage,
ScalingRequest,
ScalingRequestAlert,
)
from amqp.model.service_message_type import ServiceMessageType
from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
from amqp.router.router_base import ANY_RECIPIENT
@@ -31,8 +37,10 @@ from amqp.adapter.serializer import map_as_string
# Track the number of messages currently being processed
current_parallel_executions = 0
last_data_message_time = 0 # helps detect IDLE condition
last_backpressure_event_time = 0 # helps prevent flooding the system with backpressure events
last_data_message_time = 0 # helps detect IDLE condition
last_backpressure_event_time = (
0 # helps prevent flooding the system with backpressure events
)
last_backpressure_event = ScalingRequestAlert.IDLE
parallel_workers = int(os.getenv("PARALLEL_WORKERS", default=0))
@@ -45,8 +53,7 @@ executor = concurrent.futures.ThreadPoolExecutor(max_workers=parallel_workers +
def collect_trace_info():
trace_map = {}
TraceContextTextMapPropagator().inject(
carrier=trace_map,
context=trace.context_api.get_current()
carrier=trace_map, context=trace.context_api.get_current()
)
return trace_map
@@ -60,15 +67,17 @@ def set_text(carrier, key, value):
class DataMessageHandler:
def __init__(self,
service_adapter: CleverThisServiceAdapter,
tracer: Tracer,
message_factory: DataMessageFactory,
service_message_factory: ServiceMessageFactory,
reply_to_exchange: AbstractRobustExchange,
rabbit_mq_client: RabbitMQClient,
amq_configuration: AMQConfiguration,
loop):
def __init__(
self,
service_adapter: CleverThisServiceAdapter,
tracer: Tracer,
message_factory: DataMessageFactory,
service_message_factory: ServiceMessageFactory,
reply_to_exchange: AbstractRobustExchange,
rabbit_mq_client: RabbitMQClient,
amq_configuration: AMQConfiguration,
loop,
):
self.service_adapter: CleverThisServiceAdapter = service_adapter
self.tracer = tracer
self.message_factory: DataMessageFactory = message_factory
@@ -82,7 +91,9 @@ 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.backpressure_monitor_window = (
amq_configuration.backpressure.time_window / 1000
) # convert to seconds
global parallel_workers
_backpressure_treshold = amq_configuration.backpressure.threshold
@@ -104,7 +115,10 @@ class DataMessageHandler:
now = time.time()
event: ScalingRequestAlert = ScalingRequestAlert.UPDATE
if now - last_backpressure_event_time >= self.backpressure_monitor_window:
if current_parallel_executions == 0 and now - last_data_message_time > self.backpressure_monitor_window:
if (
current_parallel_executions == 0
and now - last_data_message_time > self.backpressure_monitor_window
):
event = ScalingRequestAlert.IDLE
elif current_parallel_executions > parallel_workers - 2:
event = ScalingRequestAlert.OVERLOAD
@@ -112,29 +126,36 @@ class DataMessageHandler:
last_backpressure_event_time = now
last_backpressure_event = event
last_backpressure_event_time = now
logging.info(f"Backpressure event: {event}, threads: {current_parallel_executions} of {parallel_workers}")
logging.info(
f"Backpressure event: {event}, threads: {current_parallel_executions} of {parallel_workers}"
)
scaling_request: ScalingRequest = ScalingRequest(
swarm_service_id, swarm_task_id,
parallel_workers, parallel_workers - current_parallel_executions,
event
swarm_service_id,
swarm_task_id,
parallel_workers,
parallel_workers - current_parallel_executions,
event,
)
# Address the message to any adapter capable of supporting BACKPRESSURE request
scale_up_request = self.service_message_factory.of(
ServiceMessageType.BACKPRESSURE,
scaling_request.to_json(),
self.reply_to if self.reply_to else ANY_RECIPIENT
self.reply_to if self.reply_to else ANY_RECIPIENT,
)
asyncio.run_coroutine_threadsafe(
self.rabbit_mq_client.router.publish_service_message(
scale_up_request, exchange=self.rabbit_mq_client.router.cm_response_exchange
scale_up_request,
exchange=self.rabbit_mq_client.router.cm_response_exchange,
),
loop=self.loop
loop=self.loop,
)
await asyncio.sleep(self.backpressure_monitor_window) # sleep in seconds
async def inbound_data_message_callback(self, message: AbstractIncomingMessage):
threading.Thread(target=lambda: asyncio.run(self.run_callback_thread(message))).start()
threading.Thread(
target=lambda: asyncio.run(self.run_callback_thread(message))
).start()
async def run_callback_thread(self, message: AbstractIncomingMessage):
"""
@@ -151,12 +172,16 @@ class DataMessageHandler:
if amq_message.magic == DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST:
await self.run_http_request_magic(message, amq_message, self.loop)
else:
logging_error("Unknown MAGIC value: [%s], don't know how to interpret the message",
amq_message.magic)
logging_error(
"Unknown MAGIC value: [%s], don't know how to interpret the message",
amq_message.magic,
)
asyncio.run_coroutine_threadsafe(message.nack(requeue=False), self.loop)
return
else:
logging_error("######[%s] No valid AMQ Message present.", message.delivery_tag)
logging_error(
"######[%s] No valid AMQ Message present.", message.delivery_tag
)
self.record_duration(start, message.delivery_tag)
asyncio.run_coroutine_threadsafe(message.ack(multiple=False), self.loop)
@@ -164,30 +189,35 @@ class DataMessageHandler:
logging.warning("Warning: Capacity close to depleted!")
# Send a scaling request to the Management service
scaling_request: ScalingRequest = ScalingRequest(
swarm_service_id, swarm_task_id,
parallel_workers, parallel_workers - current_parallel_executions,
ScalingRequestAlert.OVERLOAD
swarm_service_id,
swarm_task_id,
parallel_workers,
parallel_workers - current_parallel_executions,
ScalingRequestAlert.OVERLOAD,
)
# Address the message to any adapter capable of supporting BACKPRESSURE request
scale_up_request = self.service_message_factory.of(
ServiceMessageType.BACKPRESSURE,
scaling_request.to_json(),
self.reply_to if self.reply_to else ANY_RECIPIENT
self.reply_to if self.reply_to else ANY_RECIPIENT,
)
asyncio.run_coroutine_threadsafe(
self.rabbit_mq_client.router.publish_service_message(
scale_up_request, exchange=self.rabbit_mq_client.router.cm_response_exchange
scale_up_request,
exchange=self.rabbit_mq_client.router.cm_response_exchange,
),
loop=_loop
loop=_loop,
)
# update / reset time-window so that the OVERLOAD is not sent too often
global last_data_message_time
last_datadeliveryTag_message_time = time.time()
async def run_http_request_magic(self,
message: AbstractIncomingMessage,
amq_message: DataMessage,
_loop: AbstractEventLoop) -> None:
async def run_http_request_magic(
self,
message: AbstractIncomingMessage,
amq_message: DataMessage,
_loop: AbstractEventLoop,
) -> None:
"""
Synchronous version of on_message method. This is a wrapper around the async on_message method.
It also counts parallel executions and handles backpressure.
@@ -200,7 +230,9 @@ class DataMessageHandler:
# Increment the counter for parallel executions
current_parallel_executions += 1
logging.info(f"PARALLEL: Current Capacity [{current_parallel_executions}] of [{parallel_workers}]")
logging.info(
f"PARALLEL: Current Capacity [{current_parallel_executions}] of [{parallel_workers}]"
)
if message.reply_to:
self.reply_to = message.reply_to
if current_parallel_executions > parallel_workers - 2:
@@ -211,20 +243,31 @@ class DataMessageHandler:
response: DataResponse = await self.service_adapter.on_message(a_message)
print(f"Result in thread: {response}")
try:
logging.info(" [*] ######[%s] AMQ Message Reply.1, To=%s, msgId=%s",
message.delivery_tag, message.reply_to, response.id)
logging.info(
" [*] ######[%s] AMQ Message Reply.1, To=%s, msgId=%s",
message.delivery_tag,
message.reply_to,
response.id,
)
await self.send_reply(message.reply_to, response)
logging.info(" [*] ######[%s] AMQ Message Reply.2, To=%s",
message.delivery_tag, message.reply_to)
logging.info(
" [*] ######[%s] AMQ Message Reply.2, To=%s",
message.delivery_tag,
message.reply_to,
)
except Exception as e:
logging_error(f"Processing message: {e}")
asyncio.run_coroutine_threadsafe(message.nack(requeue=False), loop=_loop)
asyncio.run_coroutine_threadsafe(
message.nack(requeue=False), loop=_loop
)
if isinstance(amq_message, MultipartDataMessage):
# remove the downloaded files from the output_dir
for file_id, filename in amq_message.extra_data.items():
try:
logging.info(f" [*][{message.delivery_tag}] Deleting file: {file_id} in {filename}")
logging.info(
f" [*][{message.delivery_tag}] Deleting file: {file_id} in {filename}"
)
os.remove(filename)
except OSError as e:
logging_error(f"Deleting file {filename}: {e}")
@@ -234,34 +277,50 @@ class DataMessageHandler:
logging_error(f"Request handler: {e}")
return
async def reconstitute_data_message(self, message: AbstractIncomingMessage) -> DataMessage | None:
async def reconstitute_data_message(
self, message: AbstractIncomingMessage
) -> DataMessage | None:
amq_message: DataMessage | None = None
delivery_tag = message.delivery_tag
addressing: int = message.headers.get('clevermicro.addressing', 0)
addressing: int = message.headers.get("clevermicro.addressing", 0)
if addressing == 0:
amq_message = DataMessageFactory.from_bytes(message.body)
else:
amq_message = await DataMessageFactory.from_stream(message.body, rabbitmq_url=self.rabbit_mq_url, loop=self.loop)
logging.info(f" [DBG] ######[{delivery_tag}] AMQ Message from stream: {amq_message}")
amq_message = await DataMessageFactory.from_stream(
message.body, rabbitmq_url=self.rabbit_mq_url, loop=self.loop
)
logging.info(
f" [DBG] ######[{delivery_tag}] AMQ Message from stream: {amq_message}"
)
if amq_message is not None:
extra_data = await on_message(
message=message,
json_data=amq_message.body(),
rabbitmq_url=self.rabbit_mq_url,
loop=self.loop,
output_dir=self.output_dir)
logging.info(f" [x] ######[{delivery_tag}] Multipart download result: {extra_data}")
output_dir=self.output_dir,
)
logging.info(
f" [x] ######[{delivery_tag}] Multipart download result: {extra_data}"
)
amq_message = MultipartDataMessage(amq_message, extra_data)
if amq_message:
logging.info(" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, %s:%s, traceInfo=%s",
message.delivery_tag, message.consumer_tag, amq_message.id,
amq_message.method, amq_message.path, map_as_string(amq_message.trace_info))
logging.info(
" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, %s:%s, traceInfo=%s",
message.delivery_tag,
message.consumer_tag,
amq_message.id,
amq_message.method,
amq_message.path,
map_as_string(amq_message.trace_info),
)
return amq_message
def ensure_trace_info(self, amq_message: DataMessage):
propagator = TraceContextTextMapPropagator()
context = propagator.extract(carrier=amq_message.trace_info, context=trace.context_api.get_current())
context = propagator.extract(
carrier=amq_message.trace_info, context=trace.context_api.get_current()
)
self.tracer.start_span(name="AMQ-Adapter-Consumer", context=context)
logging.info(" [*] CTX=%s", context)
# current_span = self.tracer.start_span("AMQ-Adapter.Processing.Request", parent=parent_span)
@@ -274,7 +333,11 @@ class DataMessageHandler:
def record_duration(self, start, delivery_tag):
global last_data_message_time
last_data_message_time = time.time()
logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, last_data_message_time - start)
logging.info(
" [x] ######[%s] Done, single ACK: %sms.",
delivery_tag,
last_data_message_time - start,
)
async def send_reply(self, reply_to: str, response: DataResponse):
"""
@@ -288,15 +351,27 @@ class DataMessageHandler:
pika_message: Message = Message(
body=DataResponseFactory.serialize(response),
correlation_id=response.id,
content_type=response.content_type[0]
if isinstance(response.content_type, list) else response.content_type
content_type=(
response.content_type[0]
if isinstance(response.content_type, list)
else response.content_type
),
)
logging.info(
" [*] ###### Sending Reply.1 To=%s, msgId=%s", reply_to, response.id
)
_res = asyncio.run_coroutine_threadsafe(
self.reply_to_exchange.publish(
message=pika_message, routing_key=reply_to
),
self.loop,
)
logging.info(
" [*] ###### DONE EXCH PUB(%s) Sending Reply.2 To=%s, res=%s",
self.reply_to_exchange.name,
reply_to,
_res,
)
logging.info(" [*] ###### Sending Reply.1 To=%s, msgId=%s", reply_to, response.id)
_res = asyncio.run_coroutine_threadsafe(self.reply_to_exchange.publish(
message=pika_message,
routing_key=reply_to
), self.loop)
logging.info(" [*] ###### DONE EXCH PUB(%s) Sending Reply.2 To=%s, res=%s", self.reply_to_exchange.name, reply_to, _res)
async def reply_received_callback(self, message: AbstractIncomingMessage):
"""
@@ -312,7 +387,9 @@ class DataMessageHandler:
logging.info(debug)
# TODO - figure out implementation of this part, as this is sending reply back either to a service,
# or to the user
logging.info(f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}")
logging.info(
f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}"
)
future: Future = self.outstanding.pop(reply_id, None)
# if reply is None: