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

Add IDLE backpressure detection

Browse Source
This commit is contained in:
Stanislav Hejny
2025-04-06 01:24:51 +01:00
parent f7bffbe20d
commit fc110c7175
36 changed files with 904 additions and 1050 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1 -1
View File
@@ -55,4 +55,4 @@ hs_err_pid*
amq_adapter.egg-info/
build/
unused-code/
amqp/adapter/amq_route_factory.py
+5
View File
@@ -2,7 +2,9 @@ import logging
from amqp.model.model import AMQRoute
# record separator
RS = ":"
# null route, in lieu of None value
NULL_ROUTE = {
"componentName": "",
"exchange": None,
@@ -14,6 +16,9 @@ NULL_ROUTE = {
class AMQRouteFactory:
"""
Define the methods to create and validate AMQRoute objects.
"""
@staticmethod
def from_string(data):
amqp/adapter/cleverthis_service_adapter.py
+80 -89
View File
@@ -1,35 +1,33 @@
import asyncio
"""
The layer that interfaces with the CleverThis service that this Adapter is integrating with.
Methods here will invoke the actual CleverThis Service code, which are passed as Callable object.
"""
import json
import logging
import os
import re
import sys
import threading
import traceback
from aiohttp import ClientSession, ClientResponse
from fastapi import HTTPException
from typing import Dict, List, Tuple, Callable, Coroutine, AnyStr, Any
from urllib.parse import parse_qs
from xml.etree import ElementTree
from amqp.adapter.amq_response_factory import AMQResponseFactory
from amqp.service.multipart_parser import CleverMultiPartParser
from aiohttp import ClientSession, ClientResponse
from fastapi import HTTPException
from amqp.adapter import pydantic_serializer
from amqp.adapter.data_response_factory import DataResponseFactory
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)
from amqp.model.model import DataMessage, DataResponse
from amqp.service.multipart_parser import CleverMultiPartParser
def parse_request_body(message: AMQMessage):
def parse_request_body(message: DataMessage):
"""
Parses the HTTP POST request body based on the MIME type.
Args:
message (AMQMessage): Contains raw request and MIME headers.
message (DataMessage): Contains raw request and MIME headers.
Returns:
dict: Parsed data as a dictionary.
@@ -82,53 +80,55 @@ def parse_request_body(message: AMQMessage):
raise ValueError(f"Unsupported Content-Type: {content_type}")
# ============= POST Helper function =============
async def call_cleverthis_post_put_api(
message: AMQMessage,
message: DataMessage,
args: Any,
api_method: Callable[[Any, Any], Coroutine],
success_code: int,
authenticated_user: Any
) -> AMQResponse:
) -> DataResponse:
"""
Handles POST and PUT requests, which typically contain possibly complex payload data, passed as a dictionary.
Invokes the provided API method with the given arguments and authenticated user,
and returns the response as AMQResponse.
:param message: AMQMessage containing the details of the call and the payload
and returns the response as DataResponse.
:param message: DataMessage containing the details of the call and the payload
:param args: map of the arguments to pass to the API method
:param api_method: Callable - the method to invoke. Last parameter is the authenticated user
:param success_code: HTTP code to be returned when operation suceeds
:param authenticated_user: Authenticated user. The actual class depends on the service
:return: AMQResponse class
:return: DataResponse class
"""
try:
coroutine_call: Coroutine = api_method(args, authenticated_user)
result: str = await coroutine_call
return AMQResponseFactory.of(
return DataResponseFactory.of(
message.id, success_code, message.headers.get("Content-Type", "application/json"),
result.encode('utf-8'), message.trace_info
)
except HTTPException as http_error:
tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f" [E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {http_error}")
return AMQResponseFactory.of(message.id, http_error.status_code, "text/plain",
return DataResponseFactory.of(message.id, http_error.status_code, "text/plain",
str(http_error.detail).encode('utf-8'), message.trace_info)
except Exception as error:
tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f" [E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {error}")
return AMQResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
return DataResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
# ============= GET Helper function =============
async def call_cleverthis_get_api(
message: AMQMessage,
message: DataMessage,
pattern_or_data: str | dict,
api_method: Callable[[Tuple[AnyStr | Any, ...], Any], Coroutine],
authenticated_user: Any
) -> AMQResponse:
) -> DataResponse:
"""
invokes provided Callable that is expected to be a GET API method, and returns the response as AMQResponse
:param message: AMQMessage containing the details of the call
:param pattern: If there are path variables, use this pattern to extract them
Handles GET requests, which do not contain a body, but may have path variables and extra values
in the query string.
invokes provided Callable that is expected to be a GET API method, and returns the response as DataResponse
:param message: DataMessage containing the details of the call
:param pattern_or_data: If there are path variables, use this pattern to extract them
:param api_method: Callable - the method to invoke. Last parameter is the authenticated user
:param authenticated_user: Authenticated user. The actual class depends on the service
:return:
@@ -138,25 +138,26 @@ async def call_cleverthis_get_api(
if isinstance(pattern_or_data, dict):
args = pattern_or_data
else:
match = re.search(pattern_or_data, message.path)
path, args = pydantic_serializer.parse_url_query(message.path)
match = re.search(pattern_or_data, path)
if match:
args = match.groups()
args.update(match.groupdict())
else:
_msg = f"Unexpected URL: {message.path}, expected pattern: {pattern_or_data}"
return AMQResponseFactory.of(message.id, 404, "text/plain", _msg.encode('utf-8'), message.trace_info)
return DataResponseFactory.of(message.id, 404, "text/plain", _msg.encode('utf-8'), message.trace_info)
crt: Coroutine = api_method(args, authenticated_user)
result: str = await crt
return AMQResponseFactory.of(message.id, 200, "text/plain", result.encode('utf-8'), message.trace_info)
return DataResponseFactory.of(message.id, 200, "text/plain", result.encode('utf-8'), message.trace_info)
except HTTPException as http_error:
tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f" [E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {http_error}")
return AMQResponseFactory.of(message.id, http_error.status_code, "text/plain",
return DataResponseFactory.of(message.id, http_error.status_code, "text/plain",
str(http_error.detail).encode('utf-8'), message.trace_info)
except Exception as error:
tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f" [E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {error}")
return AMQResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
return DataResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
# ============= CleverThisServiceAdapter =============
@@ -164,58 +165,55 @@ class CleverThisServiceAdapter:
"""
A base class for CleverThis service adapter that handles the incoming AMQP messages and routes them to the
appropriate CleverThisService API endpoint.
IMPORTANT: This class is not intended to be used directly. It should be subclassed for each CleverThis service
and override the methods to handle the specific API calls.
"""
def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession = None):
self.amq_configuration = amq_configuration
self.session = session
self.service_host = self.amq_configuration.dispatch.service_host
self.service_port = self.amq_configuration.dispatch.service_port
self.require_authenticated_user = self.amq_configuration.amq_adapter.require_authenticated_user
def get_content_type(self, message_headers: Dict[str, List[str]]) -> str:
"""
Pull the value of the Content-Type header from the message headers.
:param message_headers: The headers of the message.
return: The content type of the message.
"""
for header, values in message_headers.items():
if header.lower() == 'content-type':
return values[0].split(';')[0]
return 'application/json'
async def get_current_user(self, token: str):
"""
To Be Overriden in subclass for given CleverThis Service, to return the current user based on the token,
in format appropriate for the service.
:param token: The token to be used for authentication.
:return: A user object or None if the user is not authenticated.
"""
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:
async def on_message(self, message) -> DataResponse:
"""
Called when a message is received from the AMQP.
:param message:
: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
logging.info(f"PARALLEL: Current Capacity [{current_parallel_executions}]")
if current_parallel_executions > parallel_workers - 2:
logging.warning("Warning: Capacity close to depleted!")
_auth: str = message.headers.get('Authorization', '')
_auth_user: Any | None = None
if _auth == '' and DEBUG_NO_AUTH == 1:
_auth = f"Bearer DEBUG_NO_AUTH"
amq_response: DataResponse | None = None
if 'Bearer' in _auth:
try:
token = _auth.split(' ')[1]
_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:
if self.require_authenticated_user:
_auth: str = message.headers.get('Authorization', '')
if 'Bearer' in _auth:
try:
token = _auth.split(' ')[1]
_auth_user = await self.get_current_user(token)
except Exception as error:
logging.error(f"Error getting user: {error}")
if _auth_user or not self.require_authenticated_user:
method = message.method.lower()
if method == 'get':
amq_response = await self.get(message, _auth_user)
@@ -230,22 +228,21 @@ class CleverThisServiceAdapter:
else:
raise ValueError(f"Unexpected HTTP method: {message.method}")
else:
amq_response = AMQResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info)
with lock:
current_parallel_executions -= 1
amq_response = DataResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info)
return amq_response
# ==================================================================================================
# ===================== C l e v e r T h i s A P I m e t h o d s ============================
# ==================================================================================================
async def get(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
async def get(self, message: DataMessage, auth_user: Any) -> DataResponse:
"""
Handles the GET requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
:param message: message from the AMQP that contains the GET request
:param auth_user: user authenticated by the API Gateway, as UserSchem
:return: AMQResponse to be sent back to the API Gateway via AMQP
:return: DataResponse to be sent back to the API Gateway via AMQP
"""
# unmatched path - try to invoke the service directly on this path
@@ -254,19 +251,19 @@ class CleverThisServiceAdapter:
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)
return AMQResponseFactory.of(message.id, _http_resp.status, _http_resp.content_type,
return DataResponseFactory.of(message.id, _http_resp.status, _http_resp.content_type,
await _http_resp.read(),
message.trace_info)
return AMQResponseFactory.of(message.id, 404, "text/plain",
return DataResponseFactory.of(message.id, 404, "text/plain",
str("Destination location does not exists").encode('utf-8'), message.trace_info)
async def put(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
async def put(self, message: DataMessage, auth_user: Any) -> DataResponse:
"""
Handles the PUT requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
:param message: message from the AMQP that contains the PUT request
:param auth_user: user authenticated by the API Gateway, as UserSchem
:return: AMQResponse to be sent back to the API Gateway via AMQP
:return: DataResponse to be sent back to the API Gateway via AMQP
"""
return await self.handle_possible_form(
message,
@@ -278,13 +275,13 @@ class CleverThisServiceAdapter:
auth_user
)
async def post(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
async def post(self, message: DataMessage, auth_user: Any) -> DataResponse:
"""
Handles the POST requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
:param message: message from the AMQP that contains the POST request, including body and headers
:param auth_user: user authenticated by the API Gateway, as UserSchem
:return: AMQResponse to be sent back to the API Gateway via AMQP
:return: DataResponse to be sent back to the API Gateway via AMQP
"""
return await self.handle_possible_form(
message,
@@ -296,32 +293,26 @@ class CleverThisServiceAdapter:
auth_user
)
async def handle_possible_form(self, message: AMQMessage, request_coroutine, auth_user: Any) -> AMQResponse:
async def handle_possible_form(self, message: DataMessage, request_coroutine, auth_user: Any) -> DataResponse:
args = parse_request_body(message)
return await request_coroutine
async def head(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
async def head(self, message: DataMessage, auth_user: Any) -> DataResponse:
"""
Handles the HEAD requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
:param message: message from the AMQP that contains the POST request, including body and headers
:param auth_user: user authenticated by the API Gateway, as UserSchem
:return: AMQResponse to be sent back to the API Gateway via AMQP
:return: DataResponse to be sent back to the API Gateway via AMQP
"""
raise NotImplementedError
async def delete(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
async def delete(self, message: DataMessage, auth_user: Any) -> DataResponse:
"""
Handles the DELETE requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
:param message: message from the AMQP that contains the POST request, including body and headers
:param auth_user: user authenticated by the API Gateway, as UserSchem
:return: AMQResponse to be sent back to the API Gateway via AMQP
:return: DataResponse to be sent back to the API Gateway via AMQP
"""
raise NotImplementedError
def get_content_type(self, message_headers: Dict[str, List[str]]) -> str:
for header, values in message_headers.items():
if header.lower() == 'content-type':
return values[0].split(';')[0]
return 'application/json'
amqp/adapter/amq_message_factory.py → amqp/adapter/data_message_factory.py
+33 -29
View File
@@ -1,3 +1,7 @@
"""
methods to create, serialize and deserialize DataMessage objects.
"""
import base64
import json
import re
@@ -5,14 +9,14 @@ from io import BytesIO
from datetime import datetime, timezone
from typing import Dict, List
from amqp.model.model import AMQMessage
from amqp.model.model import DataMessage
from amqp.model.snowflake_id import SnowflakeId
from amqp.router.serializer import map_with_list_as_string, map_as_string, parse_map_list_string, parse_map_string
from amqp.router.utils import sanitize_as_rabbitmq_name
from amqp.adapter.file_downloader import download_buffer
class AMQMessageFactory:
class DataMessageFactory:
SNOWFLAKE_ID_BITS = 80
SEQUENCE_BITS = 12
MACHINE_ID_BITS = 24
@@ -38,7 +42,7 @@ class AMQMessageFactory:
:return: factory instance
"""
if cls._instance is None:
cls._instance = AMQMessageFactory(machine_id)
cls._instance = DataMessageFactory(machine_id)
return cls._instance
def generate_snowflake_id(self) -> SnowflakeId:
@@ -61,7 +65,7 @@ class AMQMessageFactory:
self.snowflake_sequence = 0
self.snowflake_last_timestamp = timestamp
time = (timestamp - 1000 * int(AMQMessageFactory.snowflake_epoch))
time = (timestamp - 1000 * int(DataMessageFactory.snowflake_epoch))
return SnowflakeId(
time >> (64 - self.SEQUENCE_BITS - self.MACHINE_ID_BITS),
((time << (self.SEQUENCE_BITS + self.MACHINE_ID_BITS)) | self.machine_id | self.snowflake_sequence) & 0xFFFFFFFFFFFFFFFF
@@ -74,7 +78,7 @@ class AMQMessageFactory:
path: str,
headers: Dict[str, List[str]],
message,
) -> AMQMessage:
) -> DataMessage:
"""
Encapsulate / hide the system level inputs, create the AMQ message from business relevant input
:param method: a method requested (e.g. HTTP method like GET, POST in case of REST request)
@@ -82,12 +86,12 @@ class AMQMessageFactory:
:param path: path at the service
:param headers: HTTP headers
:param message: payload
:return: AMQMessage object
:return: DataMessage object
"""
serializable_headers = headers.copy() if headers else {}
payload = base64.b64encode(message.encode('utf-8') if isinstance(message, str) else message)
trace_info = {}
return AMQMessage(
return DataMessage(
self.MAGIC_BYTE_HTTP_REQUEST,
self.generate_snowflake_id(),
method,
@@ -99,10 +103,10 @@ class AMQMessageFactory:
)
@staticmethod
def amq_message_routing_key(message: AMQMessage) -> str:
def amq_message_routing_key(message: DataMessage) -> str:
"""
Constructs the message specific routing key.
:param message: AMQMessage input
:param message: DataMessage input
:return: routing key (as string) compliant to RabbitMQ allowed character set.
"""
return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}")
@@ -115,19 +119,19 @@ class AMQMessageFactory:
:return: the message timestamp
"""
bits = _id.get_bits()
low = bits[0] >> (AMQMessageFactory.SEQUENCE_BITS + AMQMessageFactory.MACHINE_ID_BITS)
high = bits[1] << (64 - AMQMessageFactory.SEQUENCE_BITS - AMQMessageFactory.MACHINE_ID_BITS)
return (high | low) + AMQMessageFactory.snowflake_epoch * 1000
low = bits[0] >> (DataMessageFactory.SEQUENCE_BITS + DataMessageFactory.MACHINE_ID_BITS)
high = bits[1] << (64 - DataMessageFactory.SEQUENCE_BITS - DataMessageFactory.MACHINE_ID_BITS)
return (high | low) + DataMessageFactory.snowflake_epoch * 1000
@staticmethod
def to_string(message: AMQMessage) -> str:
def to_string(message: DataMessage) -> str:
"""
Human-readable representation.
:param message: AMQMessage input
:param message: DataMessage input
:return: as string
"""
return (
f"AMQMessage{{"
f"DataMessage{{"
f"id={message.id}, "
f"method='{message.method}', "
f"domain='{message.domain}', "
@@ -139,9 +143,9 @@ class AMQMessageFactory:
)
@staticmethod
def serialize(message: AMQMessage) -> bytes:
def serialize(message: DataMessage) -> bytes:
"""
Converts AMQMessage to byte array
Converts DataMessage to byte array
:param message: message to serialize
:return: messages as bytes
"""
@@ -154,24 +158,24 @@ class AMQMessageFactory:
msg_length = 2 + header_length + len(message.base64_body)
with BytesIO(bytearray(msg_length)) as baos:
baos.write(prolog_len)
baos.write(AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST.encode('utf-8'))
baos.write(DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST.encode('utf-8'))
baos.write(id_bytes)
baos.write(prolog)
baos.write(message.base64_body)
return baos.getvalue()
@staticmethod
def from_bytes(input_bytes: bytes) -> AMQMessage:
def from_bytes(input_bytes: bytes) -> DataMessage:
"""
Builds the AMQMessage from its serialized (byte array) form.
Builds the DataMessage from its serialized (byte array) form.
:param input_bytes: serialized message
:return: AMQMessage instance
:return: DataMessage instance
"""
prolog_len = (input_bytes[0] << 8) + input_bytes[1]
metadata = input_bytes[2:prolog_len + 2].decode()
pattern = re.compile(
f"{AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST}"
f"{DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST}"
r"([0-9A-Fa-f]+)\|"
r"m=([^|]*)\|"
r"d=([^|]*)\|"
@@ -183,7 +187,7 @@ class AMQMessageFactory:
match = pattern.match(metadata)
if not match:
raise ValueError(
f"AMQMessage format does not match expected format. "
f"DataMessage format does not match expected format. "
f"Input string [{metadata}] does not match pattern: {pattern.pattern}"
)
@@ -198,8 +202,8 @@ class AMQMessageFactory:
headers = parse_map_list_string(headers_str)
trace_info = parse_map_string(trace_info_str)
return AMQMessage(
AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST,
return DataMessage(
DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST,
SnowflakeId.from_hex(id_str),
method,
domain,
@@ -210,12 +214,12 @@ class AMQMessageFactory:
)
@staticmethod
async def from_stream(stream: bytes, rabbitmq_url: str, loop) -> AMQMessage | None:
async def from_stream(stream: bytes, rabbitmq_url: str, loop) -> DataMessage | None:
"""
Builds the AMQMessage from its serialized (byte array) form.
Builds the DataMessage from its serialized (byte array) form.
:param stream: serialized message
:param rabbitmq_url: RabbitMQ connection URL
:return: AMQMessage instance
:return: DataMessage instance
"""
# async implementation
message_data = json.loads(stream.decode())
@@ -223,5 +227,5 @@ class AMQMessageFactory:
if req_info:
(body, eof) = await download_buffer(loop=loop, rabbitmq_url=rabbitmq_url, queue_name=req_info)
if eof == 1:
return AMQMessageFactory.from_bytes(body)
return DataMessageFactory.from_bytes(body)
return None
amqp/adapter/amq_response_factory.py → amqp/adapter/data_response_factory.py
+24 -16
View File
@@ -1,37 +1,41 @@
"""
method to create, serialize and deserialize DataResponse message
"""
import base64
import re
from io import BytesIO
from amqp.model.model import AMQResponse
from amqp.model.model import DataResponse
from amqp.model.snowflake_id import SnowflakeId
from amqp.router.serializer import parse_map_string, map_as_string
class AMQResponseFactory:
class DataResponseFactory:
MAGIC_BYTE_HTTP_REQUEST = "A"
@staticmethod
def create_async_response_message(_id: SnowflakeId) -> AMQResponse:
def create_async_response_message(_id: SnowflakeId) -> DataResponse:
"""
Helper method to create standard OK response.
:param _id: ID
:return: Response message
"""
return AMQResponse(
return DataResponse(
str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {}
)
@staticmethod
def of(id: SnowflakeId, response_code: int, content_type: str, body: bytes, trace_info: dict[str,str]) -> AMQResponse:
def of(id: SnowflakeId, response_code: int, content_type: str, body: bytes, trace_info: dict[str,str]) -> DataResponse:
"""
Create the AMQResponse message supplying all key values
Create the DataResponse message supplying all key values
:param id: SnowflakeID
:param response_code: HTTP response code
:param content_type: MIME content type
:param body: payload
:param trace_info: Jaeger trace info (id)
:return: AMQResponseMessage object
:return: DataResponseMessage object
"""
return AMQResponse(
return DataResponse(
id=id.as_string(),
response_code=response_code,
content_type=content_type,
@@ -42,7 +46,7 @@ class AMQResponseFactory:
)
@staticmethod
def serialize(response: AMQResponse) -> bytes:
def serialize(response: DataResponse) -> bytes:
"""
Serialize the message to byte array (to be sent over RabbitMQ)
:param response: Object to serialize
@@ -57,24 +61,24 @@ class AMQResponseFactory:
msg_length = 2 + header_length + len(response.response)
with BytesIO(bytearray(msg_length)) as baos:
baos.write(prolog_len)
baos.write(AMQResponseFactory.MAGIC_BYTE_HTTP_REQUEST.encode('utf-8'))
baos.write(DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST.encode('utf-8'))
baos.write(id_bytes)
baos.write(prolog)
baos.write(base64.b64encode(response.response))
return baos.getvalue()
@staticmethod
def from_bytes(input_bytes: bytes) -> AMQResponse:
def from_bytes(input_bytes: bytes) -> DataResponse:
"""
Buil;ds the AMQResponse object from its serialized form
Buil;ds the DataResponse object from its serialized form
:param input_bytes: byte array - serialized response
:return: AMQResponse object
:return: DataResponse object
"""
prolog_len = (input_bytes[0] << 8) + input_bytes[1]
metadata = input_bytes[2:prolog_len + 2].decode()
pattern = re.compile(
f"{AMQResponseFactory.MAGIC_BYTE_HTTP_REQUEST}"
f"{DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST}"
r"([0-9A-Fa-f]+)\|"
r"r=([^|]*)\|"
r"c=([^|]*)\|"
@@ -86,7 +90,7 @@ class AMQResponseFactory:
match = pattern.match(metadata)
if not match:
raise ValueError(
f"AMQResponse format does not match expected format. "
f"DataResponse format does not match expected format. "
f"Input string [{metadata}] does not match pattern: {pattern.pattern}"
)
@@ -100,7 +104,7 @@ class AMQResponseFactory:
trace_info = parse_map_string(trace_info_str)
return AMQResponse(
return DataResponse(
id_str,
response_code,
content_type,
@@ -109,3 +113,7 @@ class AMQResponseFactory:
error_cause,
trace_info,
)
# Ensure backward compatibility
AMQResponseFactory = DataResponseFactory
amqp/adapter/file_downloader.py
+4
View File
@@ -1,3 +1,7 @@
"""
implements the file downloader for the amqp adapter, where file is sent in chunks over RabbitMQ
"""
import asyncio
import json
import os
amqp/adapter/multipart_download.py
-140
View File
@@ -1,140 +0,0 @@
import asyncio
import json
import os
import threading
from collections import defaultdict
from aio_pika.abc import HeadersType, AbstractIncomingMessage, AbstractQueue, AbstractRobustChannel
from aiormq import Channel
class MultipartDownload:
def __init__(self, headers: HeadersType, channel: AbstractRobustChannel):
self.headers = headers
self.channel = channel
self.addressing: int = headers.get('clevermicro.addressing', 0)
self.files_to_download = []
self.threads = []
# Global dictionary to track completed downloads
self.download_status = defaultdict(lambda: defaultdict(bool))
self.future = None
def ensure_directory_exists(self, filepath):
"""Ensures that the directory for the given filepath exists."""
dir_name = os.path.dirname(filepath)
if dir_name and not os.path.exists(dir_name):
os.makedirs(dir_name)
async def start_queue2(self, channel: Channel, queue_name: str, callback):
dok = await channel.queue_declare(queue=queue_name, durable=False, auto_delete=True)
await channel.basic_consume(queue=queue_name, consumer_callback=callback, no_ack=True, exclusive=True)
async def start_queue(self, queue_name: str, callback):
"""
Starts consuming from a queue.
Args:
:param queue_name: The name of the queue to consume from.
:param callback: The callback function to call when a message is received.
"""
queue: AbstractQueue = await self.channel.declare_queue(name=queue_name, durable=False, auto_delete=True)
await queue.consume(callback=callback, no_ack=True, exclusive=True)
def download_file(self, channel: Channel, file_info, message_data, output_dir):
"""
Downloads a file from RabbitMQ based on the provided file information.
Args:
channel: The RabbitMQ channel.
file_info (dict): A dictionary containing file metadata
message_data (dict): the entire message
output_dir (str): The directory where the file should be saved.
"""
filename = file_info['filename']
size = file_info['size']
stream_name = file_info['streamName']
filepath = os.path.join(output_dir, filename)
self.ensure_directory_exists(filepath)
print(f"Downloading file: {filename} (size: {size} bytes) from stream: {stream_name} to {filepath}")
bytes_received = 0
with open(filepath, 'wb') as f:
async def callback(in_message: AbstractIncomingMessage):
nonlocal bytes_received
f.write(in_message.body)
bytes_received += len(in_message.body)
# print(f"Received {len(body)} bytes for {filename}. Total: {bytes_received}/{size}") # uncomment for verbose
if bytes_received >= size:
print(f"Finished downloading {filename}")
f.close()
# Update global download status
message_req_info = json.loads(message_data.get('req-info', '{}'))
if message_req_info:
self.download_status[message_req_info['from']][filename] = True
await in_message.channel.close() # stop consuming after the file is downloaded
self.start_queue2(channel, stream_name, callback)
def on_message(self, in_message: AbstractIncomingMessage, body: bytes) -> asyncio.Future:
"""
Callback function for handling messages from the 'amq-cleverswarm' queue.
Args:
ch: The RabbitMQ channel.
method: The delivery method.
properties: The message properties.
body: The message body (a JSON string).
:param body:
"""
message_data = json.loads(body.decode())
req_info = json.loads(message_data.get('req-info', '{}')) # Handle missing 'req-info'
self.future = asyncio.Future()
if not req_info:
print("Received message without 'req-info', skipping.")
in_message.ack(False)
self.future.set_result(False)
return self.future
for file_key in req_info['files']:
files_info = req_info['files'].get(file_key, []) # Handle missing file types
self.files_to_download.extend(files_info)
if not self.files_to_download:
print("No files to download in this message.")
self.future.set_result(False)
return self.future
# Create a thread for each file download
output_dir = "downloaded_files" # You can change this
for file_info in self.files_to_download:
thread = threading.Thread(
target=self.download_file,
args=(in_message.channel, file_info, message_data, output_dir)
)
self.threads.append(thread)
thread.start()
# Wait for all download threads to complete
asyncio.run(self.wait_for_download_completion(self.threads))
# After all files are downloaded, acknowledge the original message
print("All files are being downloaded. Acknowledging the main message.")
in_message.ack(False)
print(f"Message from {in_message.routing_key} acknowledged.")
return self.future
async def wait_for_download_completion(self, threads):
"""
Waits for all download threads to complete.
Args:
threads (list): A list of threads to wait for.
"""
for thread in threads:
thread.join()
self.future.set_result(True)
amqp/adapter/pydantic_serializer.py
+4
View File
@@ -1,3 +1,7 @@
"""
This module provides serialization and deserialization functions for Pydantic models that extend BaseModel
"""
import uuid
import json
from datetime import datetime
amqp/adapter/service_message_factory.py
+28 -29
View File
@@ -1,9 +1,9 @@
import base64
import logging
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.trace_info_adapter import TraceInfoAdapter
from amqp.model.model import CleverMicroServiceMessage
from amqp.model.model import ServiceMessage
from amqp.model.service_message_type import ServiceMessageType
from amqp.model.snowflake_id import SnowflakeId
from amqp.router.serializer import parse_map_string, map_as_string
@@ -12,40 +12,40 @@ RS = "|"
SPLIT_RS = r"\|"
INVALID_MESSAGE = CleverMicroServiceMessage() # default values mean invalid message
INVALID_MESSAGE = ServiceMessage() # default values mean invalid message
class CleverMicroServiceMessageFactory:
class ServiceMessageFactory:
"""
build/serialize/deserialize CleverMicro Service Message class
"""
def __init__(self, name: str):
self.process_name = name
def to_string(self, message: CleverMicroServiceMessage) -> str:
def to_string(self, message: ServiceMessage) -> str:
"""
Convert CleverMicroServiceMessage to human-readable format
:param message: CleverMicroServiceMessage object
Convert ServiceMessage to human-readable format
:param message: ServiceMessage object
:return: as string
"""
if message is not None:
return (
f"CleverMicroServiceMessage{{id={message.id.as_string()}|type={self.format_message_type(message)}|"
f"ServiceMessage{{id={message.id.as_string()}|type={self.format_message_type(message)}|"
f"recipient={message.recipient_name}|replyTo={message.reply_to}|"
f"traceInfo={message.trace_info}|body={base64.b64encode(message.message.encode()).decode() if message.payload_type & 0x80 else message.message}}}"
)
return "null"
def of(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> CleverMicroServiceMessage:
def of(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> ServiceMessage:
"""
Builds CleverMicroServiceMessage object from relevant values.
Builds ServiceMessage object from relevant values.
:param message_type: enum Message type
:param payload: Message content, stored as-is
:param recipient_name: Recipient name (from the routing expression)
:return: CleverMicroServiceMessage object
:return: ServiceMessage object
"""
_id = AMQMessageFactory.get_instance(1).generate_snowflake_id()
return CleverMicroServiceMessage(
_id = DataMessageFactory.get_instance(1).generate_snowflake_id()
return ServiceMessage(
id=_id,
payload_type=0,
message_type=message_type,
@@ -55,16 +55,16 @@ class CleverMicroServiceMessageFactory:
message=payload
)
def as_base64(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> CleverMicroServiceMessage:
def as_base64(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> ServiceMessage:
"""
Builds CleverMicroServiceMessage object from relevant values, stores content Base64 encoded.
Builds ServiceMessage object from relevant values, stores content Base64 encoded.
:param message_type: enum Message type
:param payload: Message content, stored as Base64 string
:param recipient_name: Recipient name (from the routing expression)
:return: CleverMicroServiceMessage object
:return: ServiceMessage object
"""
_id = AMQMessageFactory.get_instance(1).generate_snowflake_id()
return CleverMicroServiceMessage(
_id = DataMessageFactory.get_instance(1).generate_snowflake_id()
return ServiceMessage(
id=_id,
payload_type=0x80,
message_type=message_type,
@@ -74,13 +74,13 @@ class CleverMicroServiceMessageFactory:
message=payload
)
def from_bytes(self, input_bytes: bytes) -> CleverMicroServiceMessage:
def from_bytes(self, input_bytes: bytes) -> ServiceMessage:
"""
Deserialize CleverMicroServiceMessage from byte array serialized form.
:param input_bytes: serialized CleverMicroServiceMessage
:return: CleverMicroServiceMessage object represented by the input.
Deserialize ServiceMessage from byte array serialized form.
:param input_bytes: serialized ServiceMessage
:return: ServiceMessage object represented by the input.
"""
logging.info(f"CleverMicroServiceMessage >>> [{input_bytes.decode()}]")
logging.info(f"ServiceMessage >>> [{input_bytes.decode()}]")
input_str_array = input_bytes.decode().split(RS)
try:
i = 0
@@ -96,7 +96,7 @@ class CleverMicroServiceMessageFactory:
trace_info = parse_map_string(input_str_array[i]) if i < len(input_str_array) else {}
i += 1
message = input_str_array[i] if i < len(input_str_array) else ""
return CleverMicroServiceMessage(
return ServiceMessage(
id=SnowflakeId.from_hex(_id),
payload_type=payload_type,
message_type=message_type,
@@ -114,10 +114,10 @@ class CleverMicroServiceMessageFactory:
)
return INVALID_MESSAGE
def serialize(self, message: CleverMicroServiceMessage) -> bytes:
def serialize(self, message: ServiceMessage) -> bytes:
"""
Serialize CleverMicroServiceMessage to byte array.
:param message: CleverMicroServiceMessage to serialize.
Serialize ServiceMessage to byte array.
:param message: ServiceMessage to serialize.
:return: serialized object.
"""
return (
@@ -126,8 +126,7 @@ class CleverMicroServiceMessageFactory:
f"{base64.b64encode(message.message.encode()).decode() if message.payload_type != 0 else message.message}"
).encode('utf-8')
def format_message_type(self, message: CleverMicroServiceMessage) -> str:
def format_message_type(self, message: ServiceMessage) -> str:
"""
Formats the Enum message type and decorates it with base64 flag at highest bit.
For compatibility with 0-based enum ordinals in Java subtract 1 from ordinal value.
amqp/adapter/test_amq_message_factory.py → amqp/adapter/test_data_message_factory.py
+11 -11
View File
@@ -1,25 +1,25 @@
from datetime import datetime, timezone
from unittest import TestCase
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.data_message_factory import DataMessageFactory
class AMQMessageFactoryTest(TestCase):
class DataMessageFactoryTest(TestCase):
def test_from_bytes(self):
raw = "\002\027A64f3b5af4d00000a4000|m=POST|d=clever-pybook.localhost|p=/debug|h={Accept=[*/*],X-Forwarded-Host=[clever-pybook.localhost:8085],User-Agent=[curl/8.7.1],X-Forwarded-Proto=[http],X-Forwarded-For=[172.18.0.1],Host=[clever-pybook.localhost:8085],Accept-Encoding=[gzip],Content-Length=[1475],X-Forwarded-Port=[8085],X-Real-Ip=[172.18.0.1],X-Forwarded-Server=[c5945bef67d1],Content-Type=[multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v]}|t={traceparent=00-bed6eb7de0d8ccf4c80b145a4bfec357-8d1f3bafc471bf0f-01}|b=TEST_ME"
msg = AMQMessageFactory.from_bytes(raw.encode("utf-8"))
msg = DataMessageFactory.from_bytes(raw.encode("utf-8"))
self.assertIsNotNone(msg.id)
self.assertEqual(msg.id.as_string(),'64F3B5AF4D00000A4000'.lower(),'SnowflakeID incorrectly parsed.')
self.assertEqual(msg.headers["Content-Type"][0], "multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v")
def test_generate_snowflake_id(self):
_id = AMQMessageFactory.get_instance(1).generate_snowflake_id()
_id = DataMessageFactory.get_instance(1).generate_snowflake_id()
self.assertEqual('00001000', _id.as_string()[-8:], 'Instance ID not set correctly')
_id = AMQMessageFactory(2).generate_snowflake_id()
_id = DataMessageFactory(2).generate_snowflake_id()
self.assertEqual('00002000', _id.as_string()[-8:], 'Instance ID not set correctly')
_f = AMQMessageFactory(127)
_f = DataMessageFactory(127)
_i0 = _f.generate_snowflake_id()
_i1 = _f.generate_snowflake_id()
_i2 = _f.generate_snowflake_id()
@@ -30,7 +30,7 @@ class AMQMessageFactoryTest(TestCase):
self.assertEqual('0007F003'.lower(), _i3.as_string()[-8:], 'Instance ID and sequence not set correctly')
def test_create_request_message(self):
_f = AMQMessageFactory(34)
_f = DataMessageFactory(34)
_req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox")
self.assertEqual("localhost", _req.domain, 'Domain not set correctly')
self.assertEqual("/test/me", _req.path, 'Context path not set correctly')
@@ -38,7 +38,7 @@ class AMQMessageFactoryTest(TestCase):
self.assertEqual(b'Quick brown fox', _req.body(), "Body not correctly decoded")
def test_serialize(self):
_f = AMQMessageFactory(34)
_f = DataMessageFactory(34)
_req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox")
_ser = _f.serialize(_req)
self.assertEqual(65, _ser[2])
@@ -50,19 +50,19 @@ class AMQMessageFactoryTest(TestCase):
self.assertEqual("v2", _deser.headers["k2"][0])
def test_amq_message_routing_key(self):
_f = AMQMessageFactory(35)
_f = DataMessageFactory(35)
_req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox")
_key = _f.amq_message_routing_key(_req)
self.assertEqual("localhost.test.me", _key, "Message routing key wrongly extracted from the message")
def test_get_timestamp_from_id(self):
_f = AMQMessageFactory(35)
_f = DataMessageFactory(35)
_timestamp = _f.get_timestamp_from_id(_f.generate_snowflake_id())
_x_timestamp = int(datetime.now(timezone.utc).timestamp() * 1000)
self.assertLessEqual(_x_timestamp - _timestamp, 1, "Timestamp in SnowflakeId differs too much from current time")
def test_to_string(self):
_f = AMQMessageFactory(35)
_f = DataMessageFactory(35)
_req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox")
_as_str = _f.to_string(_req)
self.assertGreater(_as_str.index("domain='localhost'"), 0, "Expected content not found")
amqp/adapter/test_amq_response_factory.py → amqp/adapter/test_data_response_factory.py
+10 -10
View File
@@ -1,22 +1,22 @@
from unittest import TestCase
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.amq_response_factory import AMQResponseFactory
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.data_response_factory import DataResponseFactory
from amqp.model.snowflake_id import SnowflakeId
class TestAMQResponseFactory(TestCase):
class TestDataResponseFactory(TestCase):
def test_create_async_response_message(self):
_f = AMQResponseFactory()
_g = AMQMessageFactory(111)
_f = DataResponseFactory()
_g = DataMessageFactory(111)
_resp = _f.create_async_response_message(_g.generate_snowflake_id())
self.assertEqual(200, _resp.response_code)
self.assertEqual(b"OK", _resp.response)
def test_serialize(self):
_f = AMQResponseFactory()
_g = AMQMessageFactory(111)
_f = DataResponseFactory()
_g = DataMessageFactory(111)
_snowflakeId: SnowflakeId = _g.generate_snowflake_id()
_resp = _f.create_async_response_message(_snowflakeId)
_ser = _f.serialize(_resp)
@@ -25,12 +25,12 @@ class TestAMQResponseFactory(TestCase):
self.assertEqual((_ser[0]+_ser[1]), len(_ser) - 2 - len('T0s='))
def test_from_bytes(self):
_f = AMQResponseFactory()
_g = AMQMessageFactory(111)
_f = DataResponseFactory()
_g = DataMessageFactory(111)
_snowflakeId: SnowflakeId = _g.generate_snowflake_id()
_resp = _f.create_async_response_message(_snowflakeId)
_ser = _f.serialize(_resp)
_deser = _f.from_bytes(_ser)
self.assertEqual(_snowflakeId.as_string(), _deser.id)
self.assertEqual(200, _deser.response_code)
self.assertEqual(b'OK', _deser.response)
self.assertEqual(b'OK', _deser.response)
amqp/adapter/test_pydantic_serializer.py
+36
View File
@@ -0,0 +1,36 @@
from unittest import TestCase
import pytest
from amqp.adapter.pydantic_serializer import parse_url_query
class PydanticSerializerTest(TestCase):
def test_serialize_object(self):
assert False
def test_deserialize_object(self):
assert False
def test__convert_value(self):
assert False
def test__is_uuid_string(self):
assert False
def test__is_datetime_string(self):
assert False
def test__is_int_string(self):
assert False
def test_serialize_to_json(self):
assert False
def test_parse_url_query(self):
path, args = parse_url_query("http://localhost:8080/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741?response_type=JsonFile")
assert path == "/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741"
assert args == {'response_type': 'JsonFile'}
path, args = parse_url_query("http://localhost:8080/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741")
assert path == "/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741"
assert args == {}
amqp/adapter/test_service_message_factory.py
+9 -9
View File
@@ -1,19 +1,19 @@
from unittest import TestCase
from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory, RS, INVALID_MESSAGE
from amqp.model.model import CleverMicroServiceMessage
from amqp.adapter.service_message_factory import ServiceMessageFactory, RS, INVALID_MESSAGE
from amqp.model.model import ServiceMessage
from amqp.model.service_message_type import ServiceMessageType
class CleverMicroServiceMessageTest(TestCase):
class ServiceMessageTest(TestCase):
def test_to_string(self):
_f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory('bumble-bee')
message: CleverMicroServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
_f: ServiceMessageFactory = ServiceMessageFactory('bumble-bee')
message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
message_str = _f.to_string(message)
self.assertTrue("|type=01|" in message_str)
self.assertTrue("|body=Duck" in message_str)
message: CleverMicroServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald")
message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald")
message_str = _f.to_string(message)
self.assertTrue("|type=03|" in message_str)
self.assertTrue("|body=Duck" in message_str)
@@ -22,7 +22,7 @@ class CleverMicroServiceMessageTest(TestCase):
self.assertEqual("null", nullMsg)
def test_from(self):
_f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router")
_f: ServiceMessageFactory = ServiceMessageFactory("amqp-router")
message = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
serialized = _f.serialize(message)
deserialized = _f.from_bytes(serialized)
@@ -35,7 +35,7 @@ class CleverMicroServiceMessageTest(TestCase):
self.assertEqual(invalidMsg, INVALID_MESSAGE)
def test_base64(self):
_f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router")
_f: ServiceMessageFactory = ServiceMessageFactory("amqp-router")
message = _f.as_base64(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
serialized = _f.serialize(message)
assert b"|81|" in serialized
@@ -45,7 +45,7 @@ class CleverMicroServiceMessageTest(TestCase):
assert deserialized.message == "Duck"
def test_serialize(self):
_f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router")
_f: ServiceMessageFactory = ServiceMessageFactory("amqp-router")
message = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald")
serialized = _f.serialize(message)
assert RS + "0" + str(ServiceMessageType.ROUTING_DATA.value)
amqp/config/amq_configuration.py
+6 -4
View File
@@ -25,6 +25,9 @@ class AMQAdapter:
self.service_name = config.get('CleverMicro-AMQ', self.PREFIX + 'service-name',
fallback='amq-python-adapter-' + str(self.generator_id))
self.route_mapping = config.get('CleverMicro-AMQ', self.PREFIX + 'route-mapping', fallback=None)
self.require_authenticated_user = config.get(
'CleverMicro-AMQ', self.PREFIX + 'require-authenticated-user', fallback=False
)
def adapter_prefix(self) -> str:
return f"{self.service_name}-{self.generator_id}-"
@@ -72,13 +75,12 @@ class Backpressure:
"""
cm.backpressure.threshold=20
cm.backpressure.time-window=3000
cm.backpressure.management-service-url=management-service:8080
:param config: config parser to read configuration values
"""
self.management_service_url = config.get('CleverMicro-AMQ', self.PREFIX + 'management-service-url', fallback='management-service:8080')
self.threshold = config.getint('CleverMicro-AMQ', self.PREFIX + 'threshold', fallback=20)
self.time_window = config.getint('CleverMicro-AMQ', self.PREFIX + 'time-window', fallback=3000)
self.threshold = config.getint('CleverMicro-AMQ', self.PREFIX + 'threshold', fallback=0)
# time-window is in milliseconds, to report backpressure IDLE alert
self.time_window = config.getint('CleverMicro-AMQ', self.PREFIX + 'time-window', fallback=10000)
class AMQConfiguration:
amqp/config/application.properties
+8 -4
View File
@@ -3,16 +3,20 @@
# ====================================================================
[CleverMicro-AMQ]
cm.amq-adapter.service-name=amq-adapter-python
cm.amq-adapter.is-router=false
cm.amq-adapter.generator-id=164
cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:cleverswarm.localhost.#:5:0
cm.amq-adapter.require-authenticated-user=false
cm.dispatch.use-dlq=true
cm.dispatch.use-confirms=false
cm.dispatch.amq-host=rabbitmq
cm.dispatch.amq-port=5672
cm.dispatch.amq-port-tls=5671
cm.dispatch.download-dir=/tmp/downloaded_files
# Not used for CleverSwarm nor CleverBRAG or in tight coupling mode.
# applicable only in loose coupling mode, specify the destination where to forward the REST request
#cm.dispatch.service-host=cleverswarm.localhost
#cm.dispatch.service-port=8080
#
# Define maximum number of messages of being processed in parallel before triggering backpressure OVERLOAD alert
cm.backpressure.threshold=20
cm.backpressure.time-window=3000
cm.backpressure.management-service-url=management-service:8080
amqp/config/application.properties.local
+4 -3
View File
@@ -3,7 +3,6 @@
# ====================================================================
[CleverMicro-AMQ]
cm.amq-adapter.service-name=amq-adapter-python
cm.amq-adapter.is-router=false
cm.amq-adapter.generator-id=211
cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:localhost.#:5:0
@@ -12,7 +11,9 @@ cm.dispatch.use-confirms=false
cm.dispatch.amq-host=localhost
cm.dispatch.amq-port=5672
cm.dispatch.amq-port-tls=5671
# Not used for CleverSwarm nor CleverBRAG or in tight coupling mode.
# applicable only in loose coupling mode, specify the destination where to forward the REST request
#cm.dispatch.service-host=cleverswarm.localhost
#cm.dispatch.service-port=8080
cm.backpressure.threshold=20
cm.backpressure.time-window=3000
cm.backpressure.management-service-url=management-service:8080
amqp/model/model.py
+61 -10
View File
@@ -1,4 +1,8 @@
import base64
import json
from enum import Enum
from dataclasses import dataclass
from typing import Final
from typing import List, Dict
from dataclasses import dataclass, field
@@ -139,12 +143,12 @@ class AMQRoute:
@dataclass
class AMQMessage:
class DataMessage:
"""
The class represents a structure of a message that transfers the user request to the Service.
It may also be used to make a call/request between services.
The response to this call is in format of AMQResponse class.
Please see AMQMessageFactory for code to instantiate, serialize and deserialize the class
The response to this call is in format of DataResponse class.
Please see DataMessageFactory for code to instantiate, serialize and deserialize the class
"""
magic: str
id: SnowflakeId
@@ -162,19 +166,24 @@ class AMQMessage:
return base64.b64decode(self.base64_body)
class MultipartAMQMessage(AMQMessage):
def __init__(self, message: AMQMessage, extra_data: dict):
# Ensure backward compatibility
AMQMessage = DataMessage
class MultipartDataMessage(DataMessage):
def __init__(self, message: DataMessage, extra_data: dict):
super().__init__(message.magic, message.id,
message.method, message.domain, message.path,
message.headers, message.trace_info,
message.base64_body)
self.extra_data = extra_data
@dataclass
class AMQResponse:
class DataResponse:
"""
The response message to RPC style call (note: the request is made with AMQMessage)
Please see AMQResponseFactory for code to instantiate, serialize and deserialize the class
The response message to RPC style call (note: the request is made with DataMessage)
Please see DataResponseFactory for code to instantiate, serialize and deserialize the class
"""
id: str
response_code: int
@@ -188,11 +197,15 @@ class AMQResponse:
return base64.b64decode(self.response)
# Ensure backward compatibility
AMQResponse = DataResponse
@dataclass(frozen=True)
class CleverMicroServiceMessage:
class ServiceMessage:
"""
The CleverMicro service message that is used to communicate the internal state of the AMQ channel.
In v0.1 it supports service discovery / route registration and Backpressure status notification.
In v0.2 it supports service discovery / route registration and Backpressure status notification.
"""
id: SnowflakeId = None
payload_type: int = 0
@@ -205,3 +218,41 @@ class CleverMicroServiceMessage:
def is_valid(self) -> bool:
return self.message_type is not None
class ScalingRequestAlert(Enum):
OVERLOAD = "OVERLOAD"
IDLE = "IDLE"
UPDATE = "UPDATE"
@dataclass(frozen=True)
class ScalingRequest:
service_id: str
task_id: str
max_availability: int
current_availability: int
request_type: ScalingRequestAlert
version: int = 1
def to_json(self) -> str:
"""Converts the object to a JSON string matching the Java structure"""
return json.dumps({
"version": self.version,
"serviceId": self.service_id,
"taskId": self.task_id,
"maxAvailability": self.max_availability,
"currentAvailability": self.current_availability,
"requestType": self.request_type.value # Using .value for Enum serialization
}, indent=2)
@classmethod
def from_json(cls, json_str: str) -> 'ScalingRequest':
"""Creates a ScalingRequest from a JSON string"""
data = json.loads(json_str)
return cls(
service_id=data["serviceId"],
task_id=data["taskId"],
max_availability=data["maxAvailability"],
current_availability=data["currentAvailability"],
request_type=ScalingRequestAlert(data["requestType"]) # Convert string to Enum
)
amqp/model/test_snowflake_id.py
+9 -9
View File
@@ -1,26 +1,26 @@
from unittest import TestCase
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.model.snowflake_id import SnowflakeId
class SnowflakeIdTest(TestCase):
def test_equals(self):
_id_1 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
_id_2 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
_id_1 = DataMessageFactory.get_instance(1).generate_snowflake_id()
_id_2 = DataMessageFactory.get_instance(1).generate_snowflake_id()
self.assertFalse(_id_1 == _id_2)
_id_3 = SnowflakeId(_id_1.get_bytes())
self.assertTrue(_id_1 == _id_3)
def test_lt(self):
_id_1 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
_id_2 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
_id_1 = DataMessageFactory.get_instance(1).generate_snowflake_id()
_id_2 = DataMessageFactory.get_instance(1).generate_snowflake_id()
self.assertTrue(_id_1 < _id_2)
def test_from_hex(self):
_factory = AMQMessageFactory.get_instance(1)
_factory = DataMessageFactory.get_instance(1)
_id_1 = _factory.generate_snowflake_id()
_id_1_ser = _id_1.as_string()
_id_2 = SnowflakeId.from_hex(_id_1_ser)
@@ -31,18 +31,18 @@ class SnowflakeIdTest(TestCase):
in str(context.exception))
def test_get_bytes(self):
_factory = AMQMessageFactory.get_instance(1)
_factory = DataMessageFactory.get_instance(1)
_id_1 = _factory.generate_snowflake_id()
_id_1_ser = _id_1.get_bytes()
self.assertEqual(SnowflakeId.SNOWFLAKE_ID_WIDTH, len(_id_1_ser))
def test_get_bits(self):
_factory = AMQMessageFactory.get_instance(1)
_factory = DataMessageFactory.get_instance(1)
_id_1 = _factory.generate_snowflake_id()
self.assertEqual(2, len(_id_1.get_bits()))
def test_as_string(self):
_factory = AMQMessageFactory.get_instance(1)
_factory = DataMessageFactory.get_instance(1)
_id_1 = _factory.generate_snowflake_id()
self.assertEqual(2 * SnowflakeId.SNOWFLAKE_ID_WIDTH, len(_id_1.as_string()))
amqp/model/uploaded_file.py
-21
View File
@@ -1,21 +0,0 @@
import io
class UploadedFile:
def __init__(self, filename: str = None):
self.filename = filename
self.body_b64 = io.BytesIO()
def from_bytes(input_bytes: bytes) -> UploadedFile:
"""
Builds the UploadedFile from its serialized (byte array) form.
:param input_bytes: serialized message
:return: UploadedFile instance
"""
prolog_len = (input_bytes[0] << 8) + input_bytes[1]
metadata = input_bytes[2:prolog_len + 2].decode()
body_b64 = input_bytes[prolog_len + 2:]
return UploadedFile()
amqp/rabbitmq/rabbit_mq_client.py
+174
View File
@@ -0,0 +1,174 @@
import logging
import aio_pika
from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange
from pika.exchange_type import ExchangeType
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.model.model import DataMessage, AMQRoute
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE, DLQ_EXCHANGE, AMQ_RPC_EXCHANGE
from amqp.router.router_consumer import RouterConsumer
from amqp.router.utils import sanitize_as_rabbitmq_name
class RabbitMQClient:
PREFETCH_COUNT = 1
DLQ_ARGS = {"x-dead-letter-exchange": DLQ_EXCHANGE}
def __init__(self, configuration):
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 = configuration
logging.info("*******************************************")
logging.info("******** RabbitMQProducer.init(): %s",
self.amq_configuration.dispatch.amq_host)
logging.info("*******************************************")
self.router = RouterConsumer(configuration)
# this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
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)
await self.router.init_routing_paths_consumer(self.router.channel)
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=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(" [*] RabbitMQClient.cleanup()")
self.router.cleanup() # de-register route(s)
if self.router.is_open(self.channel):
try:
self.channel.close()
if not self.connection.is_closed:
self.connection.close()
except Exception as e:
logging.error(" [E] RabbitMQClient.PreDestroy cleanup - connection close error: %s", e)
async def register_inbound_routes(self, route_mapping: str, callback):
"""
Register inbound routes for the AMQ adapter. The route mapping is a comma-separated string that contains
list of routes. A route is the mapping of domain and path to the exchange and queue name.
This method declare (and thus create) the queue and exchange, bind them together using the
regex-like routing expression given in the route.
After this the Service is listening on (consuming from) the queue for incoming messages.
"""
if self.router.is_open(self.channel):
queue_args = self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None
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)
for route in requested_routes:
try:
# Note: AMQRoute allows one of the exchange or queue to be None, in which case the default is used
queue_name = route.queue if route.queue else self.router.get_consume_queue_name()
logging.info(" [*] RabbitMQClient register route: [%s]", route)
queue: AbstractRobustQueue = await self.channel.declare_queue(name=queue_name,
durable=True,
exclusive=False,
auto_delete=False,
arguments=queue_args)
exchange_name: str = route.exchange if route.exchange else AMQ_RPC_EXCHANGE
exchange: AbstractRobustExchange = await self.channel.declare_exchange(
name=exchange_name, type=ExchangeType.topic
)
await queue.bind(exchange, route.key)
await queue.consume(callback, no_ack=False)
# Register the route with the router and publish its detail to other adapters
await self.router.register_route(
AMQRoute(
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 Exception as err:
logging.error(" [E] Callback registration INCOMPLETE, %s", err)
else:
logging.warning(" [W] Channel is null, cannot register routes.")
async def register_data_reply_callback(self, rpc_callback):
await self.reply_to_queue.consume(callback=rpc_callback, no_ack=False)
async def setup_amq_channel(self, loop):
"""
Set up the RabbitMQ connection
:return: nothing, it applies changes to instance variables
"""
self.connection = await aio_pika.connect_robust(
host=self.amq_configuration.dispatch.amq_host,
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
)
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_internal_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))
async def send_message(self, message: DataMessage):
"""
The API routine that sends a datagram message (DataMessage) to the destination. The Destination
is determined from the DataMessage metadata section, using `domain` and `path` as key.
:param message: message to send
:return: nothing (void) as submitting message to RabbitMQ exchange does not return anything
meaningful. There will be asynch ACK later that will confirm the message was sent to RMQ.
"""
try:
route = self.router.find_route_by_message(message)
if route and self.rpc_exchange:
if route.timeout <= 0:
# No response expected
pika_message: aio_pika.message.Message = aio_pika.message.Message(
body=DataMessageFactory.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
pika_message: aio_pika.message.Message = aio_pika.message.Message(
body=DataMessageFactory.serialize(message),
content_encoding='utf-8',
delivery_mode=2,
correlation_id=str(message.id),
reply_to=self.router.get_reply_to_queue_name()
)
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())
else:
logging.warning("Message not sent, no route for %s/%s", message.domain, message.path)
except Exception as err:
logging.error("Send message failed: %s", err)
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)
amqp/rabbitmq/rabbit_mq_consumer.py
-81
View File
@@ -1,81 +0,0 @@
import logging
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
from amqp.router.router_consumer import RouterConsumer
from amqp.router.utils import sanitize_as_rabbitmq_name
class RabbitMQConsumer(RabbitMQProducer):
PREFETCH_COUNT = 1
DLQ_ARGS = {"x-dead-letter-exchange": DLQ_EXCHANGE}
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)
await self.router.init_routing_paths_consumer(self.router.channel)
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=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()")
self.router.cleanup() # de-register route(s)
if self.router.is_open(self.channel):
try:
self.channel.close()
except Exception as e:
logging.error(" [E] RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e)
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
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)
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
)
await queue.bind(exchange, route.key)
await queue.consume(callback, no_ack=False)
await self.router.register_route(
AMQRoute(
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 Exception as err:
logging.error(" [E] Callback registration INCOMPLETE, %s", err)
else:
logging.warning(" [W] Channel is null, cannot register routes.")
async def register_rpc_handler(self, rpc_consumer):
await self.reply_to_queue.consume(callback=rpc_consumer,
no_ack=False
)
amqp/rabbitmq/rabbit_mq_producer.py
-104
View File
@@ -1,104 +0,0 @@
import logging
import aio_pika
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import AMQMessage, AMQRoute
from amqp.router.router_producer import RouterProducer
class RabbitMQProducer:
PREFETCH_COUNT = 1
def __init__(self, amq_configuration: AMQConfiguration, router: RouterProducer = 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",
self.amq_configuration.dispatch.amq_host)
logging.info("*******************************************")
self.router = router or RouterProducer(amq_configuration)
# this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
async def init(self, loop):
await self.setup_amq_channel(loop)
def cleanup(self):
if self.connection:
try:
self.connection.close()
except Exception as e:
logging.error("RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e)
async def setup_amq_channel(self, loop):
"""
Set up the RabbitMQ connection
:return: nothing, it applies changes to instance variables
"""
self.connection = await aio_pika.connect_robust(
host=self.amq_configuration.dispatch.amq_host,
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
)
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))
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.
:param message: message to send
:return: nothing (void) as submitting message to RabbitMQ exchange does not return anything
meaningful. There will be asynch ACK later that will confirm the message was sent to RMQ.
"""
try:
route = self.router.find_route_by_message(message)
if route and self.rpc_exchange:
if route.timeout <= 0:
# No response expected
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
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()
)
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())
else:
logging.warning("Message not sent, no route for %s/%s", message.domain, message.path)
except Exception as err:
logging.error("Send message failed: %s", err)
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)
amqp/router/router_base.py
+3 -3
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.model.model import AMQRoute, AMQMessage
from amqp.model.model import AMQRoute, DataMessage
from amqp.router.route_database import RouteDatabase
from amqp.router.utils import sanitize_as_rabbitmq_name
@@ -26,7 +26,7 @@ class RouterBase:
self.route_added_notifier: Callable[[AMQRoute], None] | None = None
self.route_removed_notifier: Callable[[AMQRoute], None] | None = None
def get_routing_key(self, message: AMQMessage) -> str:
def get_routing_key(self, message: DataMessage) -> str:
return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}")
def wrap_routes(self) -> str:
@@ -46,7 +46,7 @@ class RouterBase:
def find_route(self, domain: str, path: str) -> AMQRoute | None:
return self.route_database.find_route(domain, path)
def find_route_by_message(self, message: AMQMessage) -> AMQRoute | None:
def find_route_by_message(self, message: DataMessage) -> AMQRoute | None:
return self.route_database.find_route(message.domain, message.path)
def add_routes(self, message: str) -> None:
amqp/router/router_consumer.py
+5 -5
View File
@@ -4,7 +4,7 @@ from typing import Set
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
from amqp.model.model import ServiceMessage, AMQRoute
from amqp.model.service_message_type import ServiceMessageType
from amqp.router.router_base import CM_C_REQ_QUEUE, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, ANY_RECIPIENT
from amqp.router.router_producer import RouterProducer
@@ -61,7 +61,7 @@ class RouterConsumer(RouterProducer):
await message.ack(False)
cm_message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(message.body)
cm_message: ServiceMessage = 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(cm_message), route_mapping)
@@ -69,7 +69,7 @@ class RouterConsumer(RouterProducer):
if cm_message.is_valid() and cm_message.message_type == ServiceMessageType.ROUTING_DATA_REQUEST and route_mapping:
if cm_message.reply_to != self.amq_configuration.amq_adapter.service_name:
try:
service_message: CleverMicroServiceMessage = self.service_message_factory.of(
service_message: ServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA, route_mapping, cm_message.reply_to
)
await self.publish_service_message(service_message, self.cm_response_exchange)
@@ -90,7 +90,7 @@ class RouterConsumer(RouterProducer):
"""
try:
if self.is_open(self.channel) and route.queue:
serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of(
serviceMessage: ServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA, route.as_string(), ANY_RECIPIENT
)
await self.publish_service_message(serviceMessage, self.cm_response_exchange)
@@ -110,7 +110,7 @@ class RouterConsumer(RouterProducer):
"""
logging.info(" [DBG] RouterConsumer.removeRoute, route: %s", route.as_string())
try:
service_message: CleverMicroServiceMessage = self.service_message_factory.of(
service_message: ServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA_REMOVE,
route.as_string(),
ANY_RECIPIENT
amqp/router/router_producer.py
+8 -8
View File
@@ -13,9 +13,9 @@ from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractRob
AbstractIncomingMessage
from pika.exchange_type import ExchangeType
from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory
from amqp.adapter.service_message_factory import ServiceMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import CleverMicroServiceMessage
from amqp.model.model import ServiceMessage
from amqp.model.service_message_type import ServiceMessageType
from amqp.router.router_base import RouterBase, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, CM_C_AMQ_QUEUE, \
CM_P_RESP_QUEUE, CM_P_REPLY_TO, ANY_RECIPIENT
@@ -30,7 +30,7 @@ class RouterProducer(RouterBase):
"""
super().__init__()
self.connection: AbstractRobustConnection | None = None
self.service_message_factory = CleverMicroServiceMessageFactory(configuration.amq_adapter.service_name)
self.service_message_factory = ServiceMessageFactory(configuration.amq_adapter.service_name)
self.amq_configuration: AMQConfiguration = configuration
self.channel: AbstractRobustChannel | None = None
self.cm_request_exchange: AbstractRobustExchange | None = None
@@ -40,9 +40,9 @@ class RouterProducer(RouterBase):
self.amq_configuration.amq_adapter.service_name,
self.amq_configuration.amq_adapter.route_mapping)
async def init_routing_paths(self, channel: AbstractRobustChannel | None):
async def init_internal_routing_paths(self, channel: AbstractRobustChannel | None):
"""
* Initialize router according to Producer mode.
* Initialize internal routes for ServiceMessages.
* The logic here sets a producer and listener on CleverMicroRequest/CleverMicroResponse queues
* created to enable the service discovery.
* The Producer mode sends out routing data request on CleverMicroRequest and listens
@@ -94,7 +94,7 @@ class RouterProducer(RouterBase):
"""
# ACK message now so that it is not being redelivered
await message.ack(multiple=False)
cm_message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(message.body)
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",
message.delivery_tag,
@@ -114,7 +114,7 @@ class RouterProducer(RouterBase):
"""
logging.info(" [*] RouterProducer.requestRoutesFromAdapters, queue name: %s", CM_REQUEST_EXCHANGE)
try:
serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of(
serviceMessage: ServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT
)
if not await self.publish_service_message(serviceMessage, self.cm_request_exchange):
@@ -125,7 +125,7 @@ class RouterProducer(RouterBase):
logging.error(" [E] RouterProducer failed request routing data: %s", ioe)
async def publish_service_message(self,
message: CleverMicroServiceMessage,
message: ServiceMessage,
exchange: AbstractRobustExchange) -> bool:
"""
* Publishes a service message to the service queue and logs success log entry.
amqp/router/serializer.py
+4
View File
@@ -1,3 +1,7 @@
"""
Helper functions used to aid with (de)serialization of messages sent over AMQP.
"""
import struct
from io import BytesIO
from typing import List, Union, Dict
amqp/router/test_route_database.py
+2 -2
View File
@@ -2,7 +2,7 @@ from unittest import TestCase
import pytest
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.model.model import AMQRoute
from amqp.router.route_database import RouteDatabase
@@ -75,7 +75,7 @@ class TestRouteDatabase(TestCase):
self.assertTrue(ROUTE_2 in wrap)
def test_find_route(self):
message = AMQMessageFactory.get_instance(111).create_request_message(
message = DataMessageFactory.get_instance(111).create_request_message(
"POST", "clever3-book.localhost",
"/aa/bb?cc=d", {}, ""
)
amqp/router/test_router_base.py
+3 -3
View File
@@ -3,7 +3,7 @@ from unittest import TestCase
import pytest
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.model.model import AMQRoute
from amqp.router.router_base import RouterBase
@@ -23,7 +23,7 @@ class TestRouterBase(TestCase):
@pytest.fixture(autouse=True)
def setup_each_test(self):
self.base = RouterBase()
self.factory = AMQMessageFactory.get_instance(111)
self.factory = DataMessageFactory.get_instance(111)
if len(self.base.get_routes()) == 0:
r1 = AMQRouteFactory.from_string(ROUTE_1)
r2 = AMQRouteFactory.from_string(ROUTE_2)
@@ -79,7 +79,7 @@ class TestRouterBase(TestCase):
self.assertIsNone(_route)
def test_find_route_by_message(self):
message = AMQMessageFactory.get_instance(111).create_request_message(
message = DataMessageFactory.get_instance(111).create_request_message(
"POST", "clever3-book.localhost",
"/aa/bb?cc=d", {}, ""
)
amqp/service/amq_message_handler.py
+332
View File
@@ -0,0 +1,332 @@
import asyncio
import logging
import sys
import threading
import time
import concurrent.futures
import os
import traceback
from asyncio import Future, AbstractEventLoop
from typing import Dict
from aio_pika import 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 amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.data_response_factory import DataResponseFactory
from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter
from amqp.adapter.file_downloader import on_message
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.service_message_type import ServiceMessageType
from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
from amqp.router.router_base import ANY_RECIPIENT
from amqp.router.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_backpressure_event = ScalingRequestAlert.IDLE
# Global lock to synchronize access to the current_parallel_executions variable
lock = threading.Lock()
parallel_workers = int(os.getenv("PARALLEL_WORKERS", default=0))
swarm_service_id = os.getenv("SWARM_SERVICE_ID", default="clever-amqp-service")
swarm_task_id = os.getenv("SWARM_TASK_ID", default="python-adapter")
# Shared thread pool for parallel execution
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()
)
return trace_map
def get_default_trace_state():
return TraceState(entries={})
def set_text(carrier, key, value):
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):
self.service_adapter: CleverThisServiceAdapter = service_adapter
self.tracer = tracer
self.message_factory: DataMessageFactory = message_factory
self.service_message_factory: ServiceMessageFactory = service_message_factory
self.reply_to_exchange: AbstractRobustExchange = reply_to_exchange
self.rabbit_mq_client: RabbitMQClient = rabbit_mq_client
self.rabbit_mq_url = amq_configuration.dispatch.rabbit_mq_url
self.output_dir = amq_configuration.dispatch.download_dir
self.loop = loop
# 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
global parallel_workers
_backpressure_treshold = amq_configuration.backpressure.threshold
"""
Determine the number of parallel workers based on the backpressure threshold and PARALLEL_WORKERS env var.
If env var PARALLEL_WORKERS is not set (or set to 0), use the backpressure threshold config
as the number of parallel workers.
"""
if _backpressure_treshold > 0:
if parallel_workers <= 0:
parallel_workers = _backpressure_treshold
async def monitor_backpressure(self):
while True:
global last_data_message_time
global last_backpressure_event_time
global last_backpressure_event
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:
event = ScalingRequestAlert.IDLE
elif current_parallel_executions > parallel_workers - 2:
event = ScalingRequestAlert.OVERLOAD
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}")
scaling_request: ScalingRequest = ScalingRequest(
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
)
asyncio.run_coroutine_threadsafe(
self.rabbit_mq_client.router.publish_service_message(
scale_up_request, exchange=self.rabbit_mq_client.router.cm_response_exchange
),
loop=self.loop
)
await asyncio.sleep(self.backpressure_monitor_window) # sleep in seconds
def sync_on_message(self,
message: AbstractIncomingMessage,
amq_message: DataMessage,
loop: AbstractEventLoop) -> DataResponse:
"""
Synchronous version of on_message method. This is a wrapper around the async on_message method.
It also counts parallel executions and handles backpressure.
:param message: RabbitMQ raw message
:param amq_message: DataMessage
:param loop: Event loop
:return: DataResponse
"""
global current_parallel_executions
# Increment the counter for parallel executions
with lock:
current_parallel_executions += 1
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:
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
)
# 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
)
asyncio.run_coroutine_threadsafe(
self.rabbit_mq_client.router.publish_service_message(
scale_up_request, exchange=self.rabbit_mq_client.router.cm_response_exchange
),
loop=loop
)
# update / reset time-window so that the OVERLOAD is not sent too often
global last_data_message_time
last_data_message_time = time.time()
# Call the async on_message method. This shall also ACK the message after processing.
response: DataResponse = asyncio.run(self.service_adapter.on_message(amq_message))
try:
logging.info(" [*] ######[%s] AMQ Message Reply, To=%s, msgId=%s",
message.delivery_tag, message.reply_to, response.id)
asyncio.run_coroutine_threadsafe(self.send_reply(message.reply_to, response), loop=loop)
except Exception as e:
logging.info(f"[E] Error processing message: {e}")
tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f"[E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {e}")
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}")
os.remove(filename)
except OSError as e:
logging.error(f"Error deleting file {filename}: {e}")
with lock:
current_parallel_executions -= 1
return response
async def inbound_data_message_callback(self, message: AbstractIncomingMessage):
"""
Receives the DataMessage as bytes from RabbitMQ queue and deserializes it to the DataMessage object
Ensures Jaeger trace-id propagation.
Invokes custom handler/mapper/adapter method that corresponds to this message
:param message:
:return:
"""
start = time.time()
delivery_tag = message.delivery_tag
debug = f" [DBG] ######[{delivery_tag}] (AMQ MESSAGE), env: {message.headers}, props: {message.properties}"
logging.debug(debug)
addressing: int = message.headers.get('clevermicro.addressing', 0)
amq_message: DataMessage | None = None
if addressing == 0:
amq_message = DataMessageFactory.from_bytes(message.body)
else:
loop = asyncio.get_event_loop()
amq_message = await DataMessageFactory.from_stream(message.body, rabbitmq_url=self.rabbit_mq_url, loop=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=loop,
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 is not None:
logging.info(" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, traceInfo=%s",
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())
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)
# context_with_span = trace.set_span_in_context(current_span, context)
# logging.info(" [*] CTX2=%s", context_with_span)
# propagator.inject(context_with_span, amq_message.trace_info, self.set_text)
# logging.info(" [***] NEW TI=%s CTX=%s SCOPE=%s",
# map_as_string(amq_message.trace_info), context_with_span, context)
if amq_message.magic == DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST:
future = executor.submit(self.sync_on_message, message, amq_message, self.loop)
# response: DataResponse = asyncio.run(self.http_adapter.on_message(amq_message))
# Wait and send it back to the reply queue
#try:
# response: DataResponse = future.result() # Block until the result is available
# await self.send_reply(message.reply_to, delivery_tag, response)
#except Exception as e:
# logging.info(f"[E] Error processing message: {e}")
# tb = traceback.extract_tb(sys.exc_info()[2])[-1]
# logging.error(f"[E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {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
else:
logging.error(" [E] ######[%s] No valid AMQ Message present.", 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)
await message.ack(multiple=False)
async def send_reply(self, reply_to: str, response: DataResponse):
"""
The Service side code. When the CleverXXX service has output ready, in DataResponse object,
call this method to return the reply back to the caller
:param reply_to:
:param response:
:return:
"""
if reply_to:
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
)
await self.reply_to_exchange.publish(
message=pika_message,
routing_key=reply_to
)
async def reply_received_callback(self, message: AbstractIncomingMessage):
"""
The reply for DataMessage that we sent out earlier - must match the received response with
the original request and respond to the caller.
:param message:
:return:
"""
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
logging.info(f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}")
future: Future = self.outstanding.pop(reply_id, None)
# if reply is None:
# logging.warn(" [W] ######[%s] IGNORING REPLY. No outstanding request for ID %s",
# delivery_tag, reply_id)
# else:
# result = DataResponseFactory.from_bytes(body)
# reply.tracing_span().end()
# TODO - figure out the right response format (not needed for Clever Swarm v0.1)
# that currently runs Java version, this method here is for the future compatibility to enable
# 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: DataResponse = DataResponseFactory.from_bytes(message.body)
# Complete the Future with the response data
future.set_result(response.body())
await message.ack(multiple=False)
amqp/service/amq_service.py
+34 -33
View File
@@ -6,14 +6,14 @@ from asyncio import Future
from aio_pika.abc import AbstractRobustExchange
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory
from amqp.adapter.service_message_factory import ServiceMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import AMQMessage
from amqp.rabbitmq.rabbit_mq_consumer import RabbitMQConsumer
from amqp.service.service_message_handler import AMQServiceMessageHandler
from amqp.model.model import DataMessage
from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
from amqp.service.amq_message_handler import DataMessageHandler
from amqp.service.tracing import initialize_trace
@@ -38,18 +38,18 @@ class AMQService:
self.loop = asyncio.get_event_loop()
self.amq_configuration = amq_configuration
self.message_factory = AMQMessageFactory.get_instance(self.amq_configuration.amq_adapter.generator_id)
self.rabbit_mq_consumer = RabbitMQConsumer(self.amq_configuration)
self.message_factory = DataMessageFactory.get_instance(self.amq_configuration.amq_adapter.generator_id)
self.rabbit_mq_client = RabbitMQClient(self.amq_configuration)
self.tracer = initialize_trace(name=amq_configuration.amq_adapter.service_name)
self.amq_service_message_handler = None
self.service_message_factory = CleverMicroServiceMessageFactory(
self.amq_data_message_handler = None
self.service_message_factory = ServiceMessageFactory(
self.amq_configuration.amq_adapter.service_name
)
# self.init(loop) will initialize RabbitMQ connection
self.loop.run_until_complete(self.init(self.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()
# self.consumer_thread = asyncio.create_task(self.rabbit_mq_client.channel.start_consuming())
# self.rabbit_mq_client.request_routes()
logging.info("***********************************************")
logging.info("****** AMQ Service Init Complete *********")
@@ -57,40 +57,41 @@ class AMQService:
self.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(
exchange: AbstractRobustExchange = await self.rabbit_mq_client.init(loop)
self.amq_data_message_handler = DataMessageHandler(
service_adapter,
self.tracer,
self.message_factory,
self.service_message_factory,
exchange,
self.rabbit_mq_consumer.channel,
self.amq_configuration.dispatch.rabbit_mq_url,
output_dir=self.amq_configuration.dispatch.download_dir
self.rabbit_mq_client,
self.amq_configuration,
loop=loop
)
await self.register_routes()
await self.rabbit_mq_consumer.request_routes()
await self.rabbit_mq_client.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()
logging.info(" [DBG] AMQService.cleanup() rmqc=%s", self.rabbit_mq_client)
if self.rabbit_mq_client:
self.rabbit_mq_client.cleanup()
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)
await self.rabbit_mq_consumer.init(loop=self.loop)
if not self.rabbit_mq_client.channel or self.rabbit_mq_client.channel.is_closed:
self.rabbit_mq_client = RabbitMQClient(self.amq_configuration)
await self.rabbit_mq_client.init(loop=self.loop)
await self.register_routes()
await self.rabbit_mq_consumer.request_routes()
await self.rabbit_mq_client.request_routes()
async def register_routes(self) -> AbstractRobustExchange | None:
route_mapping = self.amq_configuration.amq_adapter.route_mapping
if AMQRouteFactory.validate(route_mapping):
try:
await self.rabbit_mq_consumer.register_inbound_routes(
route_mapping, self.amq_service_message_handler.inbound_message_callback
await self.rabbit_mq_client.register_inbound_routes(
route_mapping, self.amq_data_message_handler.inbound_data_message_callback
)
return await self.rabbit_mq_consumer.register_rpc_handler(
self.amq_service_message_handler.reply_received_callback
return await self.rabbit_mq_client.register_data_reply_callback(
self.amq_data_message_handler.reply_received_callback
)
except Exception as e:
raise RuntimeError(e)
@@ -98,16 +99,16 @@ class AMQService:
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
def send_message(self, message: DataMessage, future: Future):
self.amq_data_message_handler.outstanding[str(message.id)] = future
def remove_outstanding_future(self, message_id: str):
self.amq_service_message_handler.outstanding.pop(message_id, None)
self.amq_data_message_handler.outstanding.pop(message_id, None)
# if __name__ == "__main__":
# amq_configuration: AMQConfiguration = AMQConfiguration('../config/application.properties.local')
# amq_service: AMQService = AMQService(amq_configuration, CleverSwarmAmqpAdapter(amq_configuration))
# amq_service.rabbit_mq_consumer.channel.start_consuming()
# amq_service.rabbit_mq_client.channel.start_consuming()
# amq_service.consumer_thread.cancel()
# amq_service.rabbit_mq_consumer.connection.close()
# amq_service.rabbit_mq_client.connection.close()
amqp/service/delete_queues.py
+1 -1
View File
@@ -3,7 +3,7 @@ import requests
def delete_queues_with_prefix_http(host='localhost', port=15672, username='springuser', password='TheCleverWho', prefix='cm-'):
"""
Delete queues using RabbitMQ HTTP API.
CLI utility. Delete queues using RabbitMQ HTTP API.
"""
base_url = f"http://{host}:{port}/api/queues/%2F"
amqp/service/multipart_parser.py
+2 -2
View File
@@ -7,11 +7,11 @@ from fastapi import UploadFile
from python_multipart import multipart
from python_multipart.multipart import Field, File
from amqp.model.model import AMQMessage
from amqp.model.model import DataMessage
class CleverMultiPartParser:
def __init__(self, message: AMQMessage):
def __init__(self, message: DataMessage):
self.form_data = {}
self.is_multipart = False
self.is_json = False
amqp/service/service_message_handler.py
-221
View File
@@ -1,221 +0,0 @@
import asyncio
import logging
import sys
import time
import concurrent.futures
import os
import traceback
from asyncio import Future
from typing import Dict
from aio_pika import IncomingMessage, Message
from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange, AbstractRobustChannel
from opentelemetry import trace
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.trace import Tracer, TraceState
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.adapter.file_downloader import on_message
from amqp.model.model import AMQResponse, AMQMessage, MultipartAMQMessage
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 = {}
TraceContextTextMapPropagator().inject(
carrier=trace_map,
context=trace.context_api.get_current()
)
return trace_map
def get_default_trace_state():
return TraceState(entries={})
def set_text(carrier, key, value):
carrier[key] = value
class AMQServiceMessageHandler:
def __init__(self,
service_adapter: CleverThisServiceAdapter,
tracer: Tracer,
message_factory: AMQMessageFactory,
reply_to_exchange: AbstractRobustExchange,
channel: AbstractRobustChannel,
rabbit_mq_url: str,
output_dir: str):
self.service_adapter: CleverThisServiceAdapter = service_adapter
self.tracer = tracer
self.message_factory: AMQMessageFactory = message_factory
self.reply_to_exchange: AbstractRobustExchange = reply_to_exchange
self.channel: AbstractRobustChannel = channel
self.rabbit_mq_url = rabbit_mq_url
self.output_dir = output_dir
# Dictionary to store outstanding Futures
self.outstanding: Dict[str, asyncio.Future] = {}
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.
Invokes custom handler/mapper/adapter method that corresponds to this message
:param message:
:return:
"""
start = time.time()
delivery_tag = message.delivery_tag
debug = f" [DBG] ######[{delivery_tag}] (AMQ MESSAGE), env: {message.headers}, props: {message.properties}"
logging.debug(debug)
addressing: int = message.headers.get('clevermicro.addressing', 0)
amq_message: AMQMessage | None = None
if addressing == 0:
amq_message = AMQMessageFactory.from_bytes(message.body)
else:
loop = asyncio.get_event_loop()
amq_message = await AMQMessageFactory.from_stream(message.body, rabbitmq_url=self.rabbit_mq_url, loop=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=loop,
output_dir=self.output_dir)
logging.info(f" [x] ######[{delivery_tag}] Multipart download result: {extra_data}")
amq_message = MultipartAMQMessage(amq_message, extra_data)
if amq_message is not None:
logging.info(" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, traceInfo=%s",
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())
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)
# context_with_span = trace.set_span_in_context(current_span, context)
# logging.info(" [*] CTX2=%s", context_with_span)
# propagator.inject(context_with_span, amq_message.trace_info, self.set_text)
# logging.info(" [***] NEW TI=%s CTX=%s SCOPE=%s",
# map_as_string(amq_message.trace_info), context_with_span, context)
if amq_message.magic == AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST:
future = executor.submit(self.service_adapter.sync_on_message, amq_message)
# response: AMQResponse = asyncio.run(self.http_adapter.on_message(amq_message))
# Wait 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:
logging.info(f"[E] Error processing message: {e}")
tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f"[E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {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
else:
logging.error(" [E] ######[%s] No valid AMQ Message present.", delivery_tag)
duration = time.time() - start
logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, duration)
await message.ack(multiple=False)
def on_http_response(self, delivery, deliveryTag, channel, http_response):
"""
In loose coupling mode, receives the HTTP response to the REST request the HTTPAdapter made to the
associated CleverXXX service. Convert the response to AMQResponse message and use
RabbitMQ RPC reply-to mechanism to deliver this response back to API Gateway side AMQProducer.
:param delivery:
:param deliveryTag:
:param channel:
:param http_response:
:return:
"""
response = AMQResponse(
delivery.properties.correlation_id,
http_response.status_code,
http_response.headers.get("Content-Type"),
http_response.body,
http_response.status_message,
"",
collect_trace_info()
)
try:
self.send_reply(delivery.properties.reply_to, deliveryTag, response)
except Exception as e:
tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f"[E] Error in {tb.filename}, line {tb.lineno}, function '{tb.name}': {e}")
raise RuntimeError(e)
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 response:
:return:
"""
if reply_to:
logging.info(" [*] ######[%s] AMQ Message Reply, To=%s, msgId=%s",
delivery_tag, reply_to, response.id)
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
)
await self.reply_to_exchange.publish(
message=pika_message,
routing_key=reply_to
)
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 message:
:return:
"""
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
logging.info(f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}")
future: Future = self.outstanding.pop(reply_id, None)
# if reply is None:
# logging.warn(" [W] ######[%s] IGNORING REPLY. No outstanding request for ID %s",
# delivery_tag, reply_id)
# else:
# result = AMQResponseFactory.from_bytes(body)
# reply.tracing_span().end()
# TODO - figure out the right response format (not needed for Clever Swarm v0.1)
# that currently runs Java version, this method here is for the future compatibility to enable
# 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(message.body)
# Complete the Future with the response data
future.set_result(response.body())
await message.ack(multiple=False)
amqp/service/stream_parser.py
-59
View File
@@ -1,59 +0,0 @@
import json
import struct
import io
class StreamParser:
"""
Parses input stream that contains metadata followed by binary data.
First 4 bytes are metadata length, followed by metadata in JSON format, followed by binary data.
The StreamParser accumulates the binary data from individual chunks and writes it to a file.
This can be used to parse a stream that contains a large file upload.
"""
def __init__(self):
self.buffer = io.BytesIO() # Buffer to accumulate chunks
self.metadata_length = None # Length of the metadata
self.metadata = None # Parsed JSON metadata
self.file_writer = None # File writer for the remaining data
seek_to = 0
def process_chunk(self, chunk: bytes):
"""
Process a chunk of data and handle it according to the rules.
"""
self.buffer.write(chunk)
available_data = self.buffer.tell()
# If metadata length is not yet known, try to read it
if self.metadata_length is None and available_data >= 4:
self.buffer.seek(0)
self.metadata_length = struct.unpack('>I', self.buffer.read(4))[0]
self.seek_to = self.buffer.tell()
# If metadata length is known but metadata is not yet parsed, try to parse it
if self.metadata_length is not None and self.metadata is None:
if available_data >= 4 + self.metadata_length:
self.buffer.seek(self.seek_to)
metadata_bytes = self.buffer.read(self.metadata_length)
self.metadata = json.loads(metadata_bytes.decode('utf-8'))
self.seek_to = self.buffer.tell()
print("Metadata:", self.metadata)
# If metadata is parsed, write the remaining data to the file
if self.metadata is not None:
self.buffer.seek(self.seek_to)
remaining_data = self.buffer.read()
self.seek_to = self.buffer.tell()
if remaining_data:
if self.file_writer is None:
# Prepare to write the remaining data to a file
self.file_writer = open(self.metadata.get("filename", "output_file"), "wb")
self.file_writer.write(remaining_data)
def close(self):
"""
Close the file writer and clean up.
"""
if self.file_writer:
self.file_writer.close()
self.buffer.close()
amqp/service/test_stream_parser.py
-141
View File
@@ -1,141 +0,0 @@
import unittest
import json
import struct
import os
from unittest.mock import patch, mock_open
from amqp.service.stream_parser import StreamParser
class TestStreamParser(unittest.TestCase):
def setUp(self):
"""
Set up the StreamParser instance for testing.
"""
self.parser = StreamParser()
def tearDown(self):
"""
Clean up after each test.
"""
if hasattr(self.parser, 'file_writer') and self.parser.file_writer:
self.parser.file_writer.close()
if os.path.exists("test_output_file"):
os.remove("test_output_file")
def test_process_chunk_with_metadata_and_data(self):
"""
Test processing a chunk that contains metadata and data.
"""
# Create test data
metadata = {"filename": "test_output_file", "description": "Test file"}
metadata_json = json.dumps(metadata).encode('utf-8')
metadata_length = struct.pack('>I', len(metadata_json))
data = b"some binary data"
# Combine metadata length, metadata, and data into a single chunk
chunk = metadata_length + metadata_json + data
# Process the chunk
self.parser.process_chunk(chunk)
self.parser.close()
# Verify metadata
self.assertEqual(self.parser.metadata, metadata)
self.assertEqual(self.parser.metadata_length, len(metadata_json))
# Verify file content
with open("test_output_file", "rb") as f:
file_content = f.read()
self.assertEqual(file_content, data)
def test_process_chunk_in_parts(self):
"""
Test processing chunks in multiple parts.
"""
# Create test data
metadata = {"filename": "test_output_file", "description": "Test file"}
metadata_json = json.dumps(metadata).encode('utf-8')
metadata_length = struct.pack('>I', len(metadata_json))
data = b"some binary data"
# Split the chunk into parts
chunk1 = metadata_length # First 4 bytes (metadata length)
chunk2 = metadata_json # Metadata JSON
chunk3 = data # Remaining data
# Process chunks one by one
self.parser.process_chunk(chunk1)
self.parser.process_chunk(chunk2)
self.parser.process_chunk(chunk3)
self.parser.close()
# Verify metadata
self.assertEqual(self.parser.metadata, metadata)
self.assertEqual(self.parser.metadata_length, len(metadata_json))
# Verify file content
with open("test_output_file", "rb") as f:
file_content = f.read()
self.assertEqual(file_content, data)
def test_process_chunk_without_metadata(self):
"""
Test processing a chunk without metadata.
"""
# Create test data (only metadata length and metadata, no additional data)
metadata = {"filename": "distinct_test_output_file", "description": "Test file"}
metadata_json = json.dumps(metadata).encode('utf-8')
metadata_length = struct.pack('>I', len(metadata_json))
# Combine metadata length and metadata into a single chunk
chunk = metadata_length + metadata_json
# Process the chunk
self.parser.process_chunk(chunk)
self.parser.close()
# Verify metadata
self.assertEqual(self.parser.metadata, metadata)
self.assertEqual(self.parser.metadata_length, len(metadata_json))
# Verify no data was written to the file
self.assertFalse(os.path.exists("distinct_test_output_file"))
def test_process_chunk_with_invalid_metadata(self):
"""
Test processing a chunk with invalid metadata.
"""
# Create invalid metadata (not JSON)
metadata_length = struct.pack('>I', 10)
invalid_metadata = b"invalid_json"
# Combine metadata length and invalid metadata into a single chunk
chunk = metadata_length + invalid_metadata
# Process the chunk and expect an exception
with self.assertRaises(json.JSONDecodeError):
self.parser.process_chunk(chunk)
@patch("builtins.open", new_callable=mock_open)
def test_file_writing(self, mock_file):
"""
Test file writing functionality using a mock file.
"""
# Create test data
metadata = {"filename": "test_output_file", "description": "Test file"}
metadata_json = json.dumps(metadata).encode('utf-8')
metadata_length = struct.pack('>I', len(metadata_json))
data = b"some binary data"
# Combine metadata length, metadata, and data into a single chunk
chunk = metadata_length + metadata_json + data
# Process the chunk
self.parser.process_chunk(chunk)
# Verify the file was opened and written to
mock_file.assert_called_once_with("test_output_file", "wb")
mock_file().write.assert_called_once_with(data)
pyproject.toml
+3 -2
View File
@@ -3,11 +3,12 @@ requires = ["setuptools>=61.0", "wheel"] # Specifies build tools to use
build-backend = "setuptools.build_meta"
[tool.setuptools]
packages = ["amqp","amqp.adapter","amqp.config","amqp.model","amqp.rabbitmq","amqp.router","amqp.service"] # Explicitly list packages to include
# Explicitly list packages to include`
packages = ["amqp","amqp.adapter","amqp.config","amqp.model","amqp.rabbitmq","amqp.router","amqp.service"]
[project]
name = "amq_adapter"
version = "0.2.10"
version = "0.2.14"
description = "The Python implementation of the AMQ Adaptor for CleverMicro"
readme = "README.md"
license = { text = "Apache License 2.0" }