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

#17 - review log message, standartize format and add location info

Browse Source
This commit was merged in pull request #19.
This commit is contained in:
Stanislav Hejny
2025-04-24 02:34:11 +01:00
parent d4d633cb46
commit 88d45a9b67
14 changed files with 243 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+10
View File
@@ -95,6 +95,16 @@ to the AMQConfiguration constructor. This may change in the future, when the cen
will become available in CleverMicro, but for now please ensure correct relative path wrt the current working directory.
For the config entries, please refer to the wiki page: [AMQ Adapter Config](https://docs.cleverthis.com/en/architecture/microservices/amq-adapter-configuration)
### Environment variables
The AMQ Adapter uses the following environment variables to configure the service:
PARALLEL_WORKERS - number of parallel workers to use for processing incoming requests. This is currently mandatory value,
shall be greater than 0.
AMQ_VERBOSE_LOGGING - if set to 1, the adapter will indicate in the log next to the message itself also module name, line and function name where the message was generated
```
## Donating
amqp/adapter/backpressure_handler.py
+5 -4
View File
@@ -9,6 +9,7 @@ from asyncio import AbstractEventLoop
from aio_pika import Message
from aio_pika.abc import AbstractRobustChannel, ExchangeType
from amqp.adapter.logging_utils import logging_info, logging_warning
from amqp.model.model import ScalingRequestAlert
@@ -69,7 +70,7 @@ class BackpressureHandler:
)
async def handle_backpressure_overload_event(self):
logging.warning("Warning: Capacity close to depleted!")
logging_warning("Backpressure: Capacity close to depleted!")
# Send an Overload event to the Management service
scaling_request: ScaleRequestV1 = ScaleRequestV1(
self.swarm_service_id,
@@ -84,7 +85,7 @@ class BackpressureHandler:
self.last_data_message_time = time.time()
def _handle_backpressure_idle_event(self):
logging.warning("Warning: Capacity idle!")
logging_warning("Backpressure: Service is idle.")
# Send an Idle event to the Management service
scaling_request: ScaleRequestV1 = ScaleRequestV1(
self.swarm_service_id,
@@ -130,8 +131,8 @@ class BackpressureHandler:
correlation_id=None,
)
await self.exchange.publish(message=pika_message, routing_key="")
logging.info(
" [x] Service Message Sent to %s, msg: %s",
logging_info(
"Service Message Published to %s, msg: %s",
self.exchange.name,
str(binary_content),
)
amqp/adapter/cleverthis_service_adapter.py
+17 -19
View File
@@ -23,7 +23,7 @@ from starlette.responses import FileResponse
from amqp.adapter.data_parser import AMQDataParser
from amqp.adapter.data_response_factory import DataResponseFactory
from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
from amqp.adapter.logging_utils import logging_error
from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import DataMessage, DataResponse
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
@@ -81,9 +81,7 @@ async def _convert_file_response(
exchange=_exchange_name,
routing_key=_queue_name, # Use queue name as routing key
)
logging.debug(
f" [*] Publishing to exchange: {_exchange_name}, q: {_queue_name}"
)
logging_debug(f"Publishing to exchange: {_exchange_name}, q: {_queue_name}")
return _channel, _exchange, _queue_name
# All calls to RabbitMQ API have to be called threadsafe, because those appear to be attached to parent
@@ -191,18 +189,18 @@ class CleverThisServiceAdapter:
_auth_user: Any | None = None
_amq_response: Any | None = None
logging.debug(
f" [*] on_message: _auth_user={_auth_user}, require_authenticated_user={self.require_authenticated_user}"
logging_debug(
f"on_message: require_authenticated_user={self.require_authenticated_user}"
)
if self.require_authenticated_user:
_auth: str = message.headers.get("Authorization", "")
logging.info(f"Auth: {_auth}")
logging_debug(f"Auth: {_auth}")
if isinstance(_auth, List):
_auth = _auth[0]
if "Bearer" in _auth:
try:
token = _auth.split(" ")[1]
logging.warning(f"Token: {token}")
logging_info(f"Token: {token}")
_auth_user = await self.get_current_user(token)
except Exception as error:
logging_error(f"on_message: Error getting user: {error}")
@@ -229,8 +227,8 @@ class CleverThisServiceAdapter:
_amq_response = DataResponseFactory.of(
message.id, 401, "text/plain", b"Unauthorized", message.trace_info
)
logging.info(
f" [x] ALL DONE in on_message id:{message.id}, resp code:{_amq_response.response_code}"
logging_info(
f"ALL DONE in on_message id:{message.id}, resp code:{_amq_response.response_code}"
)
return _amq_response
@@ -250,8 +248,8 @@ class CleverThisServiceAdapter:
# unmatched path - try to invoke the service directly on this path
url = f"http://{self.service_host}:{self.service_port}{message.path}"
headers = {**message.headers, **message.trace_info}
logging.info(
f" [*] SEND HEADERS out: {', '.join(f'{k}={v}' for k, v in headers.items())}, REQ={url}"
logging_info(
f"SEND HEADERS out: {', '.join(f'{k}={v}' for k, v in headers.items())}, REQ={url}"
)
if self.session:
_http_resp: ClientResponse = await self.session.get(url, headers=headers)
@@ -368,15 +366,15 @@ class CleverThisServiceAdapter:
else:
_verified_args[arg.name] = args[arg.name]
else:
logging.info(
logging_info(
f"{api_method.__name__}: Missing required argument: {arg.name}"
)
logging.info(
f" [*] INVOKE ENDPOINT: {api_method.__name__}, ARGS: {', '.join(f'{k}={v}' for k, v in _verified_args.items())}"
logging_info(
f"INVOKE ENDPOINT: {api_method.__name__}, ARGS: {', '.join(f'{k}={v}' for k, v in _verified_args.items())}"
)
_result: Any = await api_method(**_verified_args)
logging.info(
f" [*] ENDPOINT RESULT: {api_method.__name__}, RESULT: {str(_result)}"
logging_info(
f"ENDPOINT RESULT: {api_method.__name__}, RESULT: {str(_result)}"
)
if isinstance(_result, FileResponse):
@@ -406,8 +404,8 @@ class CleverThisServiceAdapter:
)
)
)
logging.info(
f" [x] ALL DONE in call_cleverthis_api id:{message.id}, resp code:{success_code}"
logging_info(
f"ALL DONE in call_cleverthis_api id:{message.id}, resp code:{success_code}"
)
return DataResponseFactory.of(
message.id,
amqp/adapter/file_handler.py
+10 -14
View File
@@ -19,7 +19,7 @@ from aio_pika.abc import (
DeliveryMode,
)
from amqp.adapter.logging_utils import logging_error, logging_debug
from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
# Global dictionary to track completed downloads
download_locations = defaultdict(str)
@@ -180,7 +180,7 @@ class StreamingFileHandler(FileHandler):
) # if a new version was created, ensure to use that new version
file_info["filename"] = filename
logging.info(
logging_debug(
f"Downloading file: {filename} (size: {size} bytes) from stream: {stream_name}:{queue_name} to {filepath}"
)
@@ -191,7 +191,7 @@ class StreamingFileHandler(FileHandler):
_eof = IN_PROGRESS
channel: AbstractRobustChannel = await connection.channel()
queue: AbstractQueue = await channel.get_queue(queue_name, ensure=False)
logging.info(
logging_debug(
f"Awaiting all file chunks being downloaded for {filename}"
)
async with queue.iterator() as queue_iter:
@@ -230,7 +230,7 @@ class StreamingFileHandler(FileHandler):
rabbitmq_url: The RabbitMQ connection URL.
queue_name (str): The directory where the file should be saved.
"""
logging.info(f"Downloading buffer from stream: {queue_name}")
logging_debug(f"Downloading buffer from stream: {queue_name}")
if queue_name:
async def wrap_rabbit_mq_calls() -> tuple[bytearray, int]:
@@ -242,7 +242,7 @@ class StreamingFileHandler(FileHandler):
queue: AbstractQueue = await channel.get_queue(
queue_name, ensure=False
)
logging.info(
logging_debug(
f"Awaiting all buffer chunks being downloaded for {queue_name}"
)
async with queue.iterator() as queue_iter:
@@ -252,7 +252,7 @@ class StreamingFileHandler(FileHandler):
_eof = message.headers.get("eof", END_OF_FILE)
if _eof != IN_PROGRESS:
break
logging.info(f"Finished downloading buffer {queue_name}")
logging_info(f"Finished downloading buffer {queue_name}")
except Exception as e:
logging_error(f"Downloading buffer: {e}")
@@ -303,7 +303,7 @@ class StreamingFileHandler(FileHandler):
files_to_download.append(files_info)
if not files_to_download:
logging.info(
logging_info(
" [%s] No files to download in this message.", message.delivery_tag
)
# await message.ack()
@@ -312,7 +312,7 @@ class StreamingFileHandler(FileHandler):
# Create a task for each file download
_tasks = []
for file_info in files_to_download:
logging.info(
logging_info(
f" [{message.delivery_tag}] about to create task for {file_info}"
)
_task = asyncio.create_task(
@@ -321,16 +321,12 @@ class StreamingFileHandler(FileHandler):
)
)
_tasks.append(_task)
logging.info(
logging_info(
f" [{message.delivery_tag}] Waiting for all files to download...gather(*tasks)"
)
await asyncio.gather(*_tasks) # Wait for all downloads to complete
# After all files are downloaded, acknowledge the original message
logging.info(f"All files downloaded. d-tag: {message.delivery_tag}")
# logging.info(f"All files downloaded. Acknowledging the main message d-tag: {message.delivery_tag}")
# await message.ack()
# logging.info(f"Message d-tag: {message.delivery_tag} acknowledged.")
logging_info(f"All files downloaded. d-tag: {message.delivery_tag}")
except Exception as e:
logging_error(f"Building DataMessage: {e}")
asyncio.run_coroutine_threadsafe(
amqp/adapter/logging_utils.py
+89 -8
View File
@@ -1,7 +1,42 @@
import inspect
import logging
import os
import sys
import traceback
verbose = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1"
def get_context_info():
"""
Retrieves the current module, line number, and function name.
Returns:
A tuple containing (module name, line number, function name).
Returns (None, None, None) if it fails to retrieve the information.
"""
try:
# Get the frame object for the current function call
frame = inspect.currentframe()
if frame is not None:
# Get the frame object for the caller of the current function
frame = frame.f_back.f_back
if frame:
code_context = frame.f_code
module_name = code_context.co_filename
module_name = (
"amqp" + module_name.split("/amqp", 1)[-1]
if "/amqp" in module_name
else module_name
)
line_number = frame.f_lineno
function_name = code_context.co_name
return module_name, line_number, function_name
else:
return None, None, None
except Exception:
return None, None, None
def logging_error(msg: str, *args) -> None:
"""
@@ -14,11 +49,58 @@ def logging_error(msg: str, *args) -> None:
if _exec_info:
_tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(
f" [E] '{msg}' ---> '{_tb.filename}': {_tb.lineno}, func: '{_tb.name}'",
f" [E] '{msg}' ---> '{_tb.filename}':{_tb.lineno}, '{_tb.name}()'",
*args,
)
else:
logging.error(f" [E] {msg}", *args)
module, line_number, function_name = get_context_info()
if module and line_number and function_name:
logging.error(
f" [E] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
*args,
)
else:
logging.error(f" [E] {msg}", *args)
def logging_warning(msg: str, *args) -> None:
"""
Log a warning message according to current logging for 'WARN' level.
Add module name, line and function name where the print occurred, on single line.
Args:
msg (str): The warning message to log.
"""
if not verbose:
logging.warning(f" [W] {msg}", *args)
return
module, line_number, function_name = get_context_info()
if module and line_number and function_name:
logging.warning(
f" [W] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
*args,
)
else:
logging.warning(f" [W] {msg}", *args)
def logging_info(msg: str, *args) -> None:
"""
Log an info message according to current logging for 'INFO' level.
Add module name, line and function name where the print occurred, on single line.
Args:
msg (str): The info message to log.
"""
if not verbose:
logging.info(f" [*] {msg}", *args)
return
module, line_number, function_name = get_context_info()
if module and line_number and function_name:
logging.info(
f" [*] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
*args,
)
else:
logging.info(f" [*] {msg}", *args)
def logging_debug(msg: str, *args) -> None:
@@ -26,13 +108,12 @@ def logging_debug(msg: str, *args) -> None:
Log a debug message according to current logging for 'DEBUG' level.
Add module name, line and function name where the print occurred, on single line.
Args:
msg (str): The error message to log.
msg (str): The debug message to log.
"""
_exec_info = sys.exc_info()[2]
if _exec_info:
_tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.debug(
f" [DBG] '{msg}' ---> '{_tb.filename}': {_tb.lineno}, func: '{_tb.name}'",
module, line_number, function_name = get_context_info()
if module and line_number and function_name:
logging.info(
f" [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
*args,
)
else:
amqp/adapter/service_message_factory.py
+2 -2
View File
@@ -2,7 +2,7 @@ import base64
import logging
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.logging_utils import logging_error
from amqp.adapter.logging_utils import logging_error, logging_info
from amqp.adapter.trace_info_adapter import TraceInfoAdapter
from amqp.model.model import ServiceMessage
from amqp.model.service_message_type import ServiceMessageType
@@ -86,7 +86,7 @@ class ServiceMessageFactory:
:param input_bytes: serialized ServiceMessage
:return: ServiceMessage object represented by the input.
"""
logging.info(f"ServiceMessage >>> [{input_bytes.decode()}]")
logging_info(f"ServiceMessage >>> [{input_bytes.decode()}]")
input_str_array = input_bytes.decode().split(RS)
try:
i = 0
amqp/config/amq_configuration.py
+4 -4
View File
@@ -3,7 +3,7 @@ import inspect
import logging
import os
from amqp.adapter.logging_utils import logging_error
from amqp.adapter.logging_utils import logging_error, logging_warning
"""
Convert configuration from properties file to an object.
@@ -138,9 +138,9 @@ class AMQConfiguration:
config_file_path = os.path.join(caller_dir, config_file_name)
config.read(config_file_path)
else:
logging_error(f" [E] Configuration file not found: {config_file_name}")
logging.warning(
" [W] Defaulting to preset values, which will likely be wrong, causing errors."
logging_error(f"Configuration file not found: {config_file_name}")
logging_warning(
"Defaulting to preset values, which will likely be wrong, causing errors."
)
# Override values with environment variables
amqp/rabbitmq/rabbit_mq_client.py
+24 -19
View File
@@ -5,7 +5,12 @@ from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange
from pika.exchange_type import ExchangeType
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.logging_utils import logging_error
from amqp.adapter.logging_utils import (
logging_error,
logging_debug,
logging_info,
logging_warning,
)
from amqp.model.model import DataMessage, AMQRoute
from amqp.router.router_base import (
AMQ_REPLY_TO_EXCHANGE,
@@ -28,7 +33,7 @@ class RabbitMQClient:
self.amq_configuration = configuration
logging.info("*******************************************")
logging.info(
"******** RabbitMQProducer.init(): %s",
"******** RabbitMQClient.init(): %s",
self.amq_configuration.dispatch.amq_host,
)
logging.info("*******************************************")
@@ -58,7 +63,7 @@ class RabbitMQClient:
return self.reply_to_exchange
def cleanup(self):
logging.info(" [*] RabbitMQClient.cleanup()")
logging_info("RabbitMQClient.cleanup()")
self.router.cleanup() # de-register route(s)
if self.router.is_open(self.channel):
try:
@@ -67,7 +72,7 @@ class RabbitMQClient:
self.connection.close()
except Exception as e:
logging_error(
" [E] RabbitMQClient.PreDestroy cleanup - connection close error: %s",
"RabbitMQClient.PreDestroy cleanup - connection close error: %s",
e,
)
@@ -83,8 +88,8 @@ class RabbitMQClient:
queue_args = (
self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None
)
logging.info(
" [*] RabbitMQClient register routes, cfg mapping: [%s]", route_mapping
logging_info(
"RabbitMQClient register routes, cfg mapping: [%s]", route_mapping
)
# convert the comma-separated string into list of AMQRoute objects
requested_routes = self.router.unwrap_route_list(route_mapping)
@@ -96,7 +101,7 @@ class RabbitMQClient:
if route.queue
else self.router.get_consume_queue_name()
)
logging.info(" [*] RabbitMQClient register route: [%s]", route)
logging_info("RabbitMQClient register route: [%s]", route)
queue: AbstractRobustQueue = await self.channel.declare_queue(
name=queue_name,
durable=True,
@@ -125,7 +130,7 @@ class RabbitMQClient:
route.valid_until,
)
)
logging.info(
logging_info(
" [%s] AMQ connection initialized for route: %s, listens on: %s/%s",
_c_tag,
route,
@@ -135,11 +140,11 @@ class RabbitMQClient:
except Exception as err:
logging_error("Callback registration INCOMPLETE, %s", err)
else:
logging.warning(" [W] Channel is null, cannot register routes.")
logging_warning("Channel is null, cannot register routes.")
async def register_data_reply_callback(self, rpc_callback):
_c_tag = await self.reply_to_queue.consume(callback=rpc_callback, no_ack=False)
logging.info(
logging_info(
" [%s] RabbitMQClient register data reply callback on queue: %s",
_c_tag,
self.reply_to_queue.name,
@@ -157,16 +162,16 @@ class RabbitMQClient:
password=self.amq_configuration.dispatch.rabbit_mq_password,
loop=loop,
)
logging.info(
"RabbitMQProducer.setupAMQ, got connection, open=%s",
logging_debug(
"RabbitMQClient.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_internal_routing_paths(self.channel)
self.router.set_route_added_notifier(
lambda route: logging.info(
" [DBG] ---------- RouteAdded, setting up metrics: amqp.adapter.exchange.%s",
lambda route: logging_debug(
" RouteAdded, setting up metrics: amqp.adapter.exchange.%s",
route.exchange,
)
)
@@ -195,8 +200,8 @@ class RabbitMQClient:
message
), # context path is a routing key
)
logging.info(
f" [x] basicPublish (no-reply), messageId: {message.id}, to: {route}"
logging_info(
f"Msg Publish (no-reply), messageId: {message.id}, to: {route}"
)
else:
# This is RPC call, there should be response
@@ -213,13 +218,13 @@ class RabbitMQClient:
message
), # context path is a routing key
)
logging.info(
" [x] basicPublish (RPC call), messageId: %s, to: %s",
logging_info(
"Msg Publish (RPC call), messageId: %s, to: %s",
message.id.as_string(),
route.as_string(),
)
else:
logging.warning(
logging_warning(
"Message not sent, no route for %s/%s", message.domain, message.path
)
amqp/router/route_database.py
+2 -1
View File
@@ -2,6 +2,7 @@ import logging
import re
from typing import Set
from amqp.adapter.logging_utils import logging_info
from amqp.model.model import AMQRoute
from amqp.router.utils import sanitize_as_rabbitmq_name
@@ -53,7 +54,7 @@ class RouteDatabase:
def find_route(self, domain: str, path: str) -> AMQRoute | None:
routing_key = sanitize_as_rabbitmq_name(f"{domain}.{path}")
logging.info(f" [DBG] findRoute key:{routing_key} in:{self.wrap_routes()}")
logging_info(f"findRoute key:{routing_key} in:{self.wrap_routes()}")
for route in self.routes:
if self.matches(route.key, routing_key):
return route
amqp/router/router_base.py
+7 -7
View File
@@ -2,7 +2,7 @@ import logging
from typing import Callable, List, Set
from amqp.adapter.amq_route_factory import AMQRouteFactory, NULL_ROUTE
from amqp.adapter.logging_utils import logging_error
from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
from amqp.model.model import AMQRoute, DataMessage
from amqp.router.route_database import RouteDatabase
from amqp.router.utils import sanitize_as_rabbitmq_name
@@ -37,7 +37,7 @@ class RouterBase:
return ",".join(str(route) for route in self.own_routes)
def unwrap_route_list(self, route_string: str) -> List[AMQRoute]:
logging.info(" [DBG] RouterMaster.unwrapRoutes, route(s): %s", route_string)
logging_debug("RouterMaster.unwrapRoutes, route(s): %s", route_string)
routes = [
AMQRouteFactory.from_string(route_str)
for route_str in route_string.split(",")
@@ -56,11 +56,11 @@ class RouterBase:
def add_routes(self, message: str) -> None:
try:
amq_route_list = self.unwrap_route_list(message)
logging.info(
logging_info(
" [DBG] RouterMaster.addRoute [%d]: %s", len(amq_route_list), message
)
for route in amq_route_list:
logging.info(" [*] RouterMaster.addRoute, adding route: %s", route)
logging_info("RouterMaster.addRoute, adding route: %s", route)
if not route.exchange and not route.queue and not route.key:
logging_error(
"Cannot setup AMQ Exchange for '%s', at least one of "
@@ -87,11 +87,11 @@ class RouterBase:
for route_str in message.split(",")
]
removed_count = sum(1 for route in to_remove if self.remove_route(route))
logging.info(
f" [*] RouterMaster.removeRoutes(): {message}, removed={removed_count}"
logging_info(
f"RouterMaster.removeRoutes(): {message}, removed={removed_count}"
)
except IOError as err:
logging.info(f" [E] Can't remove route(s): {err}")
logging_error(f"Can't remove route(s): {err}")
def remove_route(self, route: AMQRoute) -> bool:
is_route_removed = self.route_database.remove_route(route)
amqp/router/router_consumer.py
+28 -21
View File
@@ -7,7 +7,12 @@ from aio_pika.abc import (
AbstractIncomingMessage,
)
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.adapter.logging_utils import logging_error
from amqp.adapter.logging_utils import (
logging_error,
logging_info,
logging_debug,
logging_warning,
)
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import ServiceMessage, AMQRoute
from amqp.model.service_message_type import ServiceMessageType
@@ -36,14 +41,14 @@ class RouterConsumer(RouterProducer):
def __init__(self, configuration: AMQConfiguration):
super().__init__(configuration)
self.amqRouteFactory = AMQRouteFactory()
logging.info(" [*] RouterConsumer.<init>")
logging_info("RouterConsumer.<init>")
async def init_routing_paths_consumer(self, _channel: AbstractRobustChannel):
"""
* Initialize router according to Consumer mode.
"""
logging.info(
" [*] RouterConsumer.initConsumerRouter, channel open: %s",
logging_info(
"RouterConsumer.initConsumerRouter, channel open: %s",
self.is_open(_channel),
)
if self.is_open(_channel):
@@ -64,7 +69,7 @@ class RouterConsumer(RouterProducer):
_c_tag = await _queue.consume(
callback=self.handle_routing_data_request_message, no_ack=False
)
logging.info(
logging_info(
" [%s] RouterConsumer listens for RouteDataRequests on: %s",
_c_tag,
_cm_request_queue,
@@ -86,8 +91,8 @@ class RouterConsumer(RouterProducer):
# some debug statements, TODO - remove for prod release
delivery_tag = message.delivery_tag
debug = "Delivery.Properties = " + message.properties.__str__()
logging.info(
" [DBG] ******[%s] (ROUTING_DATA_REQ) tag: %s, env: %s, props: %s",
logging_debug(
" ******[%s] (ROUTING_DATA_REQ) tag: %s, env: %s, props: %s",
delivery_tag,
message.consumer_tag,
message.properties,
@@ -100,8 +105,8 @@ class RouterConsumer(RouterProducer):
message.body
)
route_mapping: str = self.wrap_own_routes()
logging.info(
" [DBG] ******[%s] (ROUTING_DATA_REQ) msg=%s, routeMap=[%s]",
logging_debug(
"******[%s] (ROUTING_DATA_REQ) msg=%s, routeMap=[%s]",
delivery_tag,
self.service_message_factory.to_string(cm_message),
route_mapping,
@@ -130,8 +135,10 @@ class RouterConsumer(RouterProducer):
err,
)
else:
logging.info(
" [*] ******[%s] (ROUTING_DATA_REQ) [%s] Ignore request originated by ourselves."
logging_info(
"******[%s] (ROUTING_DATA_REQ) [%s] Ignore request originated by ourselves.",
delivery_tag,
str(message.body),
)
else:
logging_error(
@@ -155,10 +162,10 @@ class RouterConsumer(RouterProducer):
serviceMessage, self.cm_response_exchange
)
self.add_own_route(route)
logging.info(" [DBG] RouterConsumer registered Route: %s", route)
logging_info("RouterConsumer registered Route: %s", route)
else:
logging.warning(
" [W] RouterConsumer couldn't register route data: %s channel isn't open.",
logging_warning(
"RouterConsumer couldn't register route data: %s channel isn't open.",
CM_RESPONSE_EXCHANGE,
)
except Exception as ioe:
@@ -170,7 +177,7 @@ class RouterConsumer(RouterProducer):
* :param route: route to remove
* :return result of remove operation
"""
logging.info(" [DBG] RouterConsumer.removeRoute, route: %s", route.as_string())
logging_info("RouterConsumer.removeRoute, route: %s", route.as_string())
try:
service_message: ServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA_REMOVE, route.as_string(), ANY_RECIPIENT
@@ -178,8 +185,8 @@ class RouterConsumer(RouterProducer):
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.",
logging_warning(
"RouterConsumer couldn't remove current route: %s, channel not open.",
route,
)
@@ -194,8 +201,8 @@ class RouterConsumer(RouterProducer):
* Cleanup - de-register routes with master.
"""
routes: Set[AMQRoute] = self.get_routes()
logging.info(
" [DBG] RouterConsumer.cleanup() [%d]: %s", len(routes), self.wrap_routes()
logging_info(
"RouterConsumer.cleanup() [%d]: %s", len(routes), self.wrap_routes()
)
not_removed = sum(
1
@@ -203,7 +210,7 @@ class RouterConsumer(RouterProducer):
if not r
)
if not_removed > 0:
logging.info(
" [W] %d route(s) not removed. self may lead to failed delivery on self route",
logging_warning(
"%d route(s) not removed. self may lead to failed delivery on self route",
not_removed,
)
amqp/router/router_producer.py
+23 -24
View File
@@ -19,7 +19,12 @@ from aio_pika.abc import (
)
from pika.exchange_type import ExchangeType
from amqp.adapter.logging_utils import logging_error
from amqp.adapter.logging_utils import (
logging_error,
logging_info,
logging_debug,
logging_warning,
)
from amqp.adapter.service_message_factory import ServiceMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import ServiceMessage
@@ -52,8 +57,8 @@ class RouterProducer(RouterBase):
self.cm_request_exchange: AbstractRobustExchange | None = None
self.cm_response_exchange: AbstractRobustExchange | None = None
logging.info(
f" [*] RouterProducer.<init>: %s, route: %s",
logging_info(
f"RouterProducer.<init>: %s, route: %s",
self.amq_configuration.amq_adapter.service_name,
self.amq_configuration.amq_adapter.route_mapping,
)
@@ -68,8 +73,8 @@ class RouterProducer(RouterBase):
"""
self.channel = channel
logging.info(
" [*] RouterProducer.initRoutingPaths, channel open: %s",
logging_info(
"RouterProducer.initRoutingPaths, channel open: %s",
self.is_open(self.channel),
)
if self.is_open(self.channel):
@@ -102,7 +107,7 @@ class RouterProducer(RouterBase):
_c_tag = await _response_queue.consume(
callback=self.handle_routing_data_message, no_ack=False
)
logging.info(
logging_info(
" [%s] Routing master listens for Route updates on %s",
_c_tag,
_cm_response_queue,
@@ -120,25 +125,19 @@ class RouterProducer(RouterBase):
* 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 {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()))
"""
logging.info(
" [%s] ACK message now so that it is not being redelivered",
logging_info(
f"[%s] Received ROUTING DATA: [%s], msgId=%s",
message.delivery_tag,
message.body,
message.message_id,
)
await message.ack(multiple=False)
cm_message: ServiceMessage = 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",
logging_debug(
"******[%s] ROUTING_DATA, msg=%s, routeMap=[%s] sn=%s, replyTo=%s",
message.delivery_tag,
self.service_message_factory.to_string(cm_message),
self.amq_configuration.amq_adapter.route_mapping,
@@ -159,8 +158,8 @@ class RouterProducer(RouterBase):
"""
* Send a broadcast message prompting all listening adapters to reply with own routing data.
"""
logging.info(
" [*] RouterProducer.requestRoutesFromAdapters, queue name: %s",
logging_info(
"RouterProducer.requestRoutesFromAdapters, queue name: %s",
CM_REQUEST_EXCHANGE,
)
try:
@@ -170,8 +169,8 @@ class RouterProducer(RouterBase):
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.",
logging_warning(
"RouterProducer could not request route data: %s channel is not open.",
CM_REQUEST_EXCHANGE,
)
@@ -201,8 +200,8 @@ class RouterProducer(RouterBase):
correlation_id=None,
)
await exchange.publish(message=pika_message, routing_key="")
logging.info(
" [x] Service Message Sent to %s, msg: %s",
logging_info(
"Service Message Published to %s, msg: %s",
exchange.name,
str(binary_content),
)
amqp/service/amq_message_handler.py
+18 -32
View File
@@ -19,7 +19,7 @@ from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.data_response_factory import DataResponseFactory
from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
from amqp.adapter.file_handler import StreamingFileHandler, FileHandler
from amqp.adapter.logging_utils import logging_error, logging_debug
from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
from amqp.adapter.service_message_factory import ServiceMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import (
@@ -144,7 +144,7 @@ class DataMessageHandler:
"""
# Increment the counter for parallel executions
BackpressureHandler.current_parallel_executions += 1
logging.info(
logging_info(
f"PARALLEL: Current Capacity [{BackpressureHandler.current_parallel_executions}] of [{BackpressureHandler.parallel_workers}]"
)
if message.reply_to:
@@ -160,18 +160,7 @@ class DataMessageHandler:
_response: DataResponse = await self.service_adapter.on_message(_a_message)
logging_debug(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,
)
await self.send_reply(message.reply_to, _response)
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)
@@ -180,8 +169,8 @@ class DataMessageHandler:
# 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:
@@ -204,8 +193,8 @@ class DataMessageHandler:
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}"
logging_info(
f"######[{delivery_tag}] AMQ Message from stream: {amq_message}"
)
if amq_message is not None:
extra_data = await self.file_handler.on_message(
@@ -215,13 +204,13 @@ class DataMessageHandler:
loop=self.loop,
output_dir=self.output_dir,
)
logging.info(
f" [x] ######[{delivery_tag}] Multipart download result: {extra_data}"
logging_info(
f"######[{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",
logging_info(
"######[%s] AMQ Message received, ctag=%s, msgId=%s, %s:%s, traceInfo=%s",
message.delivery_tag,
message.consumer_tag,
amq_message.id,
@@ -237,7 +226,7 @@ class DataMessageHandler:
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)
# logging_info("CTX=%s", context)
# current_span = self.tracer.start_span("AMQ-Adapter.Processing.Request", parent=parent_span)
# context_with_span = trace.set_span_in_context(current_span, context)
# logging.info(" [*] CTX2=%s", context_with_span)
@@ -248,8 +237,8 @@ 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.",
logging_info(
"######[%s] Done, single ACK: %sms.",
delivery_tag,
last_data_message_time - start,
)
@@ -272,17 +261,14 @@ class DataMessageHandler:
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",
logging_debug(
"###### DONE Reply Published to exchg:(%s) To=%s, res=%s",
self.reply_to_exchange.name,
reply_to,
_res,
@@ -298,11 +284,11 @@ class DataMessageHandler:
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)
debug = f"######[{delivery_tag}] (RPC REPLY) env: {message.headers}, props: {message.properties}, BODY: {message.body.decode()}"
logging_debug(debug)
# TODO - figure out implementation of this part, as this is sending reply back either to a service,
# or to the user
logging.info(
logging_info(
f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}"
)
amqp/service/amq_service.py
+4 -3
View File
@@ -8,6 +8,7 @@ from aio_pika.abc import AbstractRobustExchange
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.adapter.logging_utils import logging_info, logging_warning
from amqp.adapter.service_message_factory import ServiceMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
@@ -74,7 +75,7 @@ class AMQService:
await self.rabbit_mq_client.request_routes()
def cleanup(self):
logging.info(" [DBG] AMQService.cleanup() rmqc=%s", self.rabbit_mq_client)
logging_info("AMQService.cleanup() rmqc=%s", self.rabbit_mq_client)
if self.rabbit_mq_client:
self.rabbit_mq_client.cleanup()
@@ -99,8 +100,8 @@ class AMQService:
except Exception as e:
raise RuntimeError(e)
elif route_mapping:
logging.info(
" [W] route [%s] failed validation, no routes were registered.",
logging_warning(
"route [%s] failed validation, no routes were registered.",
route_mapping,
)
return None