From d2b1b59c66f1c312cbe6dc0cfcd79f6db3ca0ca2 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Wed, 15 Jan 2025 20:32:11 +0000 Subject: [PATCH] Converted AMQPAdapeter Java code to Python --- .gitignore | 2 + amqp/__init__.py | 0 amqp/adapter/__init__.py | 0 amqp/adapter/amq_message_factory.py | 166 ++++++++++++++++ amqp/adapter/amq_response_factory.py | 75 ++++++++ amqp/adapter/amq_route_factory.py | 60 ++++++ amqp/adapter/http_adapter.py | 68 +++++++ amqp/adapter/response_holder.py | 15 ++ amqp/adapter/service_message_factory.py | 107 +++++++++++ amqp/adapter/test_amq_message_factory.py | 68 +++++++ amqp/adapter/test_amq_response_factory.py | 36 ++++ amqp/adapter/test_amq_route_factory.py | 20 ++ amqp/adapter/test_service_message_factory.py | 46 +++++ amqp/adapter/trace_info_adapter.py | 14 ++ amqp/config/__init__.py | 0 amqp/config/amq_configuration.py | 81 ++++++++ amqp/config/application.properties | 18 ++ amqp/model/__init__.py | 0 amqp/model/model.py | 107 +++++++++++ amqp/model/snowflake_id.py | 68 +++++++ amqp/rabbitmq/__init__.py | 0 amqp/rabbitmq/rabbit_mq_consumer.py | 143 ++++++++++++++ amqp/rabbitmq/rabbit_mq_producer.py | 134 +++++++++++++ amqp/router/__init__.py | 0 amqp/router/route_database.py | 49 +++++ amqp/router/router_base.py | 91 +++++++++ amqp/router/router_consumer.py | 143 ++++++++++++++ amqp/router/router_producer.py | 190 +++++++++++++++++++ amqp/router/serializer.py | 61 ++++++ amqp/router/test_route_database.py | 24 +++ amqp/router/test_router_base.py | 39 ++++ amqp/router/test_utils.py | 15 ++ amqp/router/utils.py | 9 + amqp/service/__init__.py | 0 amqp/service/amq_service.py | 84 ++++++++ amqp/service/service_message_handler.py | 143 ++++++++++++++ amqp/service/tracing.py | 89 +++++++++ logging.conf | 34 ++++ requirements.txt | 8 + 39 files changed, 2207 insertions(+) create mode 100644 amqp/__init__.py create mode 100644 amqp/adapter/__init__.py create mode 100644 amqp/adapter/amq_message_factory.py create mode 100644 amqp/adapter/amq_response_factory.py create mode 100644 amqp/adapter/amq_route_factory.py create mode 100644 amqp/adapter/http_adapter.py create mode 100644 amqp/adapter/response_holder.py create mode 100644 amqp/adapter/service_message_factory.py create mode 100644 amqp/adapter/test_amq_message_factory.py create mode 100644 amqp/adapter/test_amq_response_factory.py create mode 100644 amqp/adapter/test_amq_route_factory.py create mode 100644 amqp/adapter/test_service_message_factory.py create mode 100644 amqp/adapter/trace_info_adapter.py create mode 100644 amqp/config/__init__.py create mode 100644 amqp/config/amq_configuration.py create mode 100644 amqp/config/application.properties create mode 100644 amqp/model/__init__.py create mode 100644 amqp/model/model.py create mode 100644 amqp/model/snowflake_id.py create mode 100644 amqp/rabbitmq/__init__.py create mode 100644 amqp/rabbitmq/rabbit_mq_consumer.py create mode 100644 amqp/rabbitmq/rabbit_mq_producer.py create mode 100644 amqp/router/__init__.py create mode 100644 amqp/router/route_database.py create mode 100644 amqp/router/router_base.py create mode 100644 amqp/router/router_consumer.py create mode 100644 amqp/router/router_producer.py create mode 100644 amqp/router/serializer.py create mode 100644 amqp/router/test_route_database.py create mode 100644 amqp/router/test_router_base.py create mode 100644 amqp/router/test_utils.py create mode 100644 amqp/router/utils.py create mode 100644 amqp/service/__init__.py create mode 100644 amqp/service/amq_service.py create mode 100644 amqp/service/service_message_handler.py create mode 100644 amqp/service/tracing.py create mode 100644 logging.conf create mode 100644 requirements.txt diff --git a/.gitignore b/.gitignore index d94278b..703b6eb 100644 --- a/.gitignore +++ b/.gitignore @@ -48,3 +48,5 @@ hs_err_pid* .project .settings/ +**/__pycache__ +.idea diff --git a/amqp/__init__.py b/amqp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/amqp/adapter/__init__.py b/amqp/adapter/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/amqp/adapter/amq_message_factory.py b/amqp/adapter/amq_message_factory.py new file mode 100644 index 0000000..fc0c786 --- /dev/null +++ b/amqp/adapter/amq_message_factory.py @@ -0,0 +1,166 @@ +import base64 +import re +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 AMQResponse +from amqp.router.utils import sanitize_as_rabbitmq_name +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 + + +class AMQMessageFactory: + SNOWFLAKE_ID_BITS = 80 + SEQUENCE_BITS = 12 + MACHINE_ID_BITS = 26 + TIMESTAMP_BITS = SNOWFLAKE_ID_BITS - SEQUENCE_BITS - MACHINE_ID_BITS + + MACHINE_ID_MASK = (1 << MACHINE_ID_BITS) - 1 + SEQUENCE_MASK = (1 << SEQUENCE_BITS) - 1 + MAGIC_BYTE_HTTP_REQUEST = "A" + + _instance = None + snowflake_epoch = datetime(2024, 10, 1, 0, 0, tzinfo=timezone.utc).timestamp() + + def __init__(self, machine_id: int): + self.machine_id = (machine_id & self.MACHINE_ID_MASK) << self.SEQUENCE_BITS + self.snowflake_sequence = 0 + self.snowflake_last_timestamp = 0 + + @classmethod + def get_instance(cls, machine_id: int): + if cls._instance is None: + cls._instance = AMQMessageFactory(machine_id) + return cls._instance + + def generate_snowflake_id(self) -> SnowflakeId: + timestamp = int(datetime.now(timezone.utc).timestamp() * 1000) + + if timestamp < self.snowflake_last_timestamp: + raise RuntimeError("Clock moved backwards. Refusing to generate id") + + if timestamp == self.snowflake_last_timestamp: + self.snowflake_sequence = (self.snowflake_sequence + 1) & self.SEQUENCE_MASK + if self.snowflake_sequence == 0: + while timestamp == self.snowflake_last_timestamp: + timestamp = int(datetime.now(timezone.utc).timestamp() * 1000) + self.snowflake_last_timestamp = timestamp + else: + self.snowflake_sequence = 0 + self.snowflake_last_timestamp = timestamp + + time = (timestamp - 1000 * int(AMQMessageFactory.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 + ) + + def create_request_message( + self, + method: str, + domain: str, + path: str, + headers: Dict[str, List[str]], + message: str, + ) -> AMQMessage: + serializable_headers = headers.copy() if headers else {} + payload = base64.b64encode(message.encode('utf-8')) + trace_info = {} + return AMQMessage( + self.MAGIC_BYTE_HTTP_REQUEST, + self.generate_snowflake_id(), + method, + domain, + path, + serializable_headers, + trace_info, + payload, + ) + + @staticmethod + def amq_message_routing_key(message: AMQMessage) -> str: + return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}") + + @staticmethod + def get_timestamp_from_id(_id: SnowflakeId) -> int: + 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 + + @staticmethod + def to_string(message: AMQMessage) -> str: + return ( + f"AMQMessage{{" + f"id={message.id}, " + f"method='{message.method}', " + f"domain='{message.domain}', " + f"path='{message.path}', " + f"headers={{{map_with_list_as_string(message.headers)}}}, " + f"traceInfo={{{map_as_string(message.trace_info)}}}, " + f"base64body={message.base64_body.decode()}" + f"}}" + ) + + @staticmethod + def serialize(message: AMQMessage) -> bytes: + id_bytes = str(message.id).encode() + prolog = ( + f"|m={message.method}|d={message.domain}|p={message.path}|h={{{map_with_list_as_string(message.headers)}}}|t={{{map_as_string(message.trace_info)}}}|b=" + ).encode('utf-8') + header_length = 1 + len(id_bytes) + len(prolog) + prolog_len = header_length.to_bytes(2, byteorder='big') + 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(id_bytes) + baos.write(prolog) + baos.write(message.base64_body) + return baos.getvalue() + + @staticmethod + def from_bytes(input_bytes: bytes) -> AMQMessage: + 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}" + r"([0-9A-Fa-f]+)\|" + r"m=([^|]*)\|" + r"d=([^|]*)\|" + r"p=([^|]*)\|" + r"h=(\{[^}]*\})\|" + r"t=(\{[^}]*\})\|" + r"b=" + ) + match = pattern.match(metadata) + if not match: + raise ValueError( + f"AMQMessage format does not match expected format. " + f"Input string [{metadata}] does not match pattern: {pattern.pattern}" + ) + + id_str = match.group(1) + method = match.group(2) + domain = match.group(3) + path = match.group(4) + headers_str = match.group(5) + trace_info_str = match.group(6) + body_b64 = input_bytes[prolog_len + 2:] + + headers = parse_map_list_string(headers_str) + trace_info = parse_map_string(trace_info_str) + + return AMQMessage( + AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST, + SnowflakeId.from_hex(id_str), + method, + domain, + path, + headers, + trace_info, + body_b64, + ) diff --git a/amqp/adapter/amq_response_factory.py b/amqp/adapter/amq_response_factory.py new file mode 100644 index 0000000..8968f11 --- /dev/null +++ b/amqp/adapter/amq_response_factory.py @@ -0,0 +1,75 @@ +import base64 +import re +from io import BytesIO +from amqp.model.model import AMQResponse +from amqp.model.snowflake_id import SnowflakeId +from amqp.router.serializer import parse_map_string, map_as_string + + +class AMQResponseFactory: + MAGIC_BYTE_HTTP_REQUEST = "A" + + @staticmethod + def create_async_response_message(_id: SnowflakeId) -> AMQResponse: + return AMQResponse( + str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {} + ) + + @staticmethod + def serialize(response: AMQResponse) -> bytes: + id_bytes = response.id.encode('utf-8') + prolog = ( + f"|r={response.response_code}|c={response.content_type}|e={response.error}|f={response.error_cause}|t={{{map_as_string(response.trace_info)}}}|b=" + ).encode('utf-8') + header_length = 1 + len(id_bytes) + len(prolog) + prolog_len = header_length.to_bytes(2, byteorder='big') + 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(id_bytes) + baos.write(prolog) + baos.write(base64.b64encode(response.response)) + return baos.getvalue() + + @staticmethod + def from_bytes(input_bytes: bytes) -> AMQResponse: + 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}" + r"([0-9A-Fa-f]+)\|" + r"r=([^|]*)\|" + r"c=([^|]*)\|" + r"e=([^|]*)\|" + r"f=([^|]*)\|" + r"t=(\{[^}]*\})\|" + r"b=" + ) + match = pattern.match(metadata) + if not match: + raise ValueError( + f"AMQResponse format does not match expected format. " + f"Input string [{metadata}] does not match pattern: {pattern.pattern}" + ) + + id_str = match.group(1) + response_code = int(match.group(2)) + content_type = match.group(3) + error = match.group(4) + error_cause = match.group(5) + trace_info_str = match.group(6) + body_b64 = input_bytes[prolog_len + 2:] + + trace_info = parse_map_string(trace_info_str) + + return AMQResponse( + id_str, + response_code, + content_type, + base64.b64decode(body_b64), + error, + error_cause, + trace_info, + ) diff --git a/amqp/adapter/amq_route_factory.py b/amqp/adapter/amq_route_factory.py new file mode 100644 index 0000000..c0f44b6 --- /dev/null +++ b/amqp/adapter/amq_route_factory.py @@ -0,0 +1,60 @@ +import logging + +from amqp.model.model import AMQRoute + +RS = ":" +NULL_ROUTE = { + "componentName": "", + "exchange": None, + "queue": None, + "routingKey": "#", + "timeout": 1, + "validUntil": 0 +} + + +class AMQRouteFactory: + + @staticmethod + def from_string(data): + try: + keys = data.split(RS) + if len(keys) >= 5: + i = 0 + component_name = keys[i] + i += 1 + exchange = keys[i] + i += 1 + queue = keys[i] if i < len(keys) else "" + i += 1 + routing_key = keys[i] if i < len(keys) else "" + i += 1 + timeout = int(keys[i]) if i < len(keys) else 0 + i += 1 + valid_until = int(keys[i]) if i < len(keys) else 0 + return AMQRoute( + componentName=component_name, + exchange=exchange, + queue=queue, + key=routing_key, + timeout=timeout, + validUntil=valid_until + ) + logging.error( + " [E] AMQRoute incorrect format. AMQRoute expects input as: " + "'Component-Name:AMQExchange-Name:Queue-Name:Routing-Key:Timeout:Valid-Until', " + "got: ['%s']", data + ) + except Exception as err: + logging.error( + " [E] %s: AMQRoute incorrect format. AMQRoute expects input as: " + "'Component-Name:AMQExchange-Name:Queue-Name:Routing-Key:Timeout:Valid-Until', " + "got: ['%s']", err, data + ) + return NULL_ROUTE + + @staticmethod + def validate(route_str): + if route_str is not None: + return route_str.count(":") > 4 + return False diff --git a/amqp/adapter/http_adapter.py b/amqp/adapter/http_adapter.py new file mode 100644 index 0000000..aba9f48 --- /dev/null +++ b/amqp/adapter/http_adapter.py @@ -0,0 +1,68 @@ +import logging +from typing import Dict, List +from aiohttp import ClientSession, ClientResponse + +from amqp.config.amq_configuration import AMQConfiguration +from amqp.model.model import AMQMessage + + +class HttpAdapter: + def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession): + 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 + + async def handle(self, message: AMQMessage) -> ClientResponse: + method = message.method.lower() + if method == 'get': + return await self.get(message) + elif method == 'put': + return await self.put(message) + elif method == 'post': + return await self.post(message) + elif method == 'head': + return await self.head(message) + elif method == 'delete': + return await self.delete(message) + else: + raise ValueError(f"Unexpected HTTP method: {message.method}") + + async def get(self, message: AMQMessage) -> ClientResponse: + url = f"http://{self.service_host}:{self.service_port}{message.path}" + headers = {**message.headers, **message.trace_info} + logging.info(f" [*] SEND HEADERS out: {', '.join(f'{k}={v}' for k, v in headers.items())}, REQ={url}") + return await self.session.get(url, headers=headers) + + async def put(self, message: AMQMessage) -> ClientResponse: + return await self.handle_possible_form( + self.session.put( + f"http://{self.service_host}:{self.service_port}{message.path}", + data=message.body, + headers={**message.headers, **message.trace_info}, + ) + ) + + async def post(self, message: AMQMessage) -> ClientResponse: + return await self.handle_possible_form( + self.session.post( + f"http://{self.service_host}:{self.service_port}{message.path}", + data=message.body, + headers={**message.headers, **message.trace_info}, + ) + ) + + async def handle_possible_form(self, request_coroutine) -> ClientResponse: + return await request_coroutine + + async def head(self, message: AMQMessage) -> ClientResponse: + raise NotImplementedError + + async def delete(self, message: AMQMessage) -> ClientResponse: + 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' diff --git a/amqp/adapter/response_holder.py b/amqp/adapter/response_holder.py new file mode 100644 index 0000000..1e5538b --- /dev/null +++ b/amqp/adapter/response_holder.py @@ -0,0 +1,15 @@ +from dataclasses import dataclass +from concurrent.futures import Future + +from aiohttp.web_response import Response +from opentelemetry.trace import Span + + +@dataclass +class ResponseHolder: + response: Future[Response] + tracing_span: Span + + def __init__(self, response: Future[Response], span: Span): + self.response = response + self.tracing_span = span diff --git a/amqp/adapter/service_message_factory.py b/amqp/adapter/service_message_factory.py new file mode 100644 index 0000000..74d8645 --- /dev/null +++ b/amqp/adapter/service_message_factory.py @@ -0,0 +1,107 @@ +import base64 +import logging + +from amqp.adapter.amq_message_factory import AMQMessageFactory +from amqp.adapter.trace_info_adapter import TraceInfoAdapter +from amqp.model.model import CleverMicroServiceMessage, ServiceMessageType +from amqp.model.snowflake_id import SnowflakeId +from amqp.router.serializer import parse_map_string, map_as_string + +RS = "|" +SPLIT_RS = r"\|" + + +INVALID_MESSAGE = CleverMicroServiceMessage() # default values mean invalid message + + +class CleverMicroServiceMessageFactory: + + def __init__(self, name: str): + self.process_name = name + + def to_string(self, message: CleverMicroServiceMessage) -> str: + if message is not None: + return ( + f"CleverMicroServiceMessage{{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: + _id = AMQMessageFactory.get_instance(1).generate_snowflake_id() + return CleverMicroServiceMessage( + id=_id, + payload_type=0, + message_type=message_type, + recipient_name=recipient_name, + reply_to=self.process_name, + trace_info=TraceInfoAdapter.create_produce_span(_id), + message=payload + ) + + def as_base64(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> CleverMicroServiceMessage: + _id = AMQMessageFactory.get_instance(1).generate_snowflake_id() + return CleverMicroServiceMessage( + id=_id, + payload_type=0x80, + message_type=message_type, + recipient_name=recipient_name, + reply_to=self.process_name, + trace_info=TraceInfoAdapter.create_produce_span(_id.as_string()), + message=payload + ) + + def from_bytes(self, input_bytes: bytes) -> CleverMicroServiceMessage: + print(f"CleverMicroServiceMessage >>> [{input_bytes.decode()}]") + input_str_array = input_bytes.decode().split(RS) + try: + i = 0 + _id = input_str_array[i] + i += 1 + payload_type = int(input_str_array[i], 16) + i += 1 + message_type = ServiceMessageType(payload_type + 1 & 0x7F) + recipient_name = input_str_array[i] if i < len(input_str_array) else "" + i += 1 + reply_to = input_str_array[i] if i < len(input_str_array) else "" + i += 1 + 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( + id=SnowflakeId.from_hex(_id), + payload_type=payload_type, + message_type=message_type, + recipient_name=recipient_name, + reply_to=reply_to, + trace_info=trace_info, + message=base64.b64decode(message).decode() if payload_type & 0x80 else message + ) + except Exception as e: + logging.error( + "Unexpected ServiceMessage format. Expected 6 tokens separated by '|' in format: " + "[ID(64bit number)|MessageType(8bit number)|RecipientName(String)|" + "ReplyTo(String)|TraceInfo|Payload(String or Base64)], got: " + "[{}], reported problem: {}", input_bytes.decode(), str(e) + ) + return INVALID_MESSAGE + + def serialize(self, message: CleverMicroServiceMessage) -> bytes: + return ( + f"{message.id.as_string()}{RS}{self.format_message_type(message)}{RS}{message.recipient_name}{RS}" + f"{message.reply_to}{RS}{map_as_string(message.trace_info)}{RS}" + 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: + """ + Formats the Enum message type and ecorates it with base64 flag at highest bit. + For compatibility with 0-based enum ordinals in Java subtract 1 from ordinal value. + :param message: Service message to send + :return: formatted message type + """ + num_type = (message.message_type.value if message.message_type else ServiceMessageType.UNKNOWN.value) - 1 + fmt = ((num_type & 0x7F) | (message.payload_type & 0x80)) + return f"{fmt:02X}" diff --git a/amqp/adapter/test_amq_message_factory.py b/amqp/adapter/test_amq_message_factory.py new file mode 100644 index 0000000..8a782c0 --- /dev/null +++ b/amqp/adapter/test_amq_message_factory.py @@ -0,0 +1,68 @@ +from datetime import datetime, timezone +from unittest import TestCase + +from amqp.adapter.amq_message_factory import AMQMessageFactory + + +class AMQMessageFactoryTest(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")) + self.assertIsNotNone(msg.id) + self.assertEqual(msg.id.as_string(),'64F3B5AF4D00000A4000','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() + self.assertEqual('00001000', _id.as_string()[-8:], 'Instance ID not set correctly') + + _id = AMQMessageFactory(2).generate_snowflake_id() + self.assertEqual('00002000', _id.as_string()[-8:], 'Instance ID not set correctly') + + _f = AMQMessageFactory(127) + _i0 = _f.generate_snowflake_id() + _i1 = _f.generate_snowflake_id() + _i2 = _f.generate_snowflake_id() + _i3 = _f.generate_snowflake_id() + self.assertEqual('0007F000', _i0.as_string()[-8:], 'Instance ID and sequence not set correctly') + self.assertEqual('0007F001', _i1.as_string()[-8:], 'Instance ID and sequence not set correctly') + self.assertEqual('0007F002', _i2.as_string()[-8:], 'Instance ID and sequence not set correctly') + self.assertEqual('0007F003', _i3.as_string()[-8:], 'Instance ID and sequence not set correctly') + + def test_create_request_message(self): + _f = AMQMessageFactory(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') + self.assertEqual("v1", _req.headers['k1'][0], "Headers not set correctly") + self.assertEqual(b'Quick brown fox', _req.body(), "Body not correctly decoded") + + def test_serialize(self): + _f = AMQMessageFactory(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]) + _deser = _f.from_bytes(_ser) + self.assertEqual(_req.id.as_string(), _deser.id.as_string()) + self.assertEqual("localhost", _deser.domain) + self.assertEqual("/test/me", _deser.path) + self.assertEqual(b"Quick brown fox", _deser.body()) + self.assertEqual("v2", _deser.headers["k2"][0]) + + def test_amq_message_routing_key(self): + _f = AMQMessageFactory(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) + _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) + _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") \ No newline at end of file diff --git a/amqp/adapter/test_amq_response_factory.py b/amqp/adapter/test_amq_response_factory.py new file mode 100644 index 0000000..ec8b515 --- /dev/null +++ b/amqp/adapter/test_amq_response_factory.py @@ -0,0 +1,36 @@ +from unittest import TestCase + +from amqp.adapter.amq_message_factory import AMQMessageFactory +from amqp.adapter.amq_response_factory import AMQResponseFactory +from amqp.model.snowflake_id import SnowflakeId + + +class TestAMQResponseFactory(TestCase): + + def test_create_async_response_message(self): + _f = AMQResponseFactory() + _g = AMQMessageFactory(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) + _snowflakeId: SnowflakeId = _g.generate_snowflake_id() + _resp = _f.create_async_response_message(_snowflakeId) + _ser = _f.serialize(_resp) + self.assertEqual(2, str(_ser[2:]).index(_f.MAGIC_BYTE_HTTP_REQUEST+_snowflakeId.as_string()), "Magic byte and SnowflakeId not found") + self.assertGreater(str(_ser[2:]).index("|e=|"), 2) + self.assertEqual((_ser[0]+_ser[1]), len(_ser) - 2 - len('T0s=')) + + def test_from_bytes(self): + _f = AMQResponseFactory() + _g = AMQMessageFactory(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) \ No newline at end of file diff --git a/amqp/adapter/test_amq_route_factory.py b/amqp/adapter/test_amq_route_factory.py new file mode 100644 index 0000000..c7c9aac --- /dev/null +++ b/amqp/adapter/test_amq_route_factory.py @@ -0,0 +1,20 @@ +from unittest import TestCase + +from amqp.adapter.amq_route_factory import AMQRouteFactory, NULL_ROUTE +from amqp.model.model import AMQRoute + + +class TestAMQRouteFactory(TestCase): + def test_from_string(self): + _f = AMQRouteFactory() + _r: AMQRoute = _f.from_string("amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0") + self.assertEqual("", _r.exchange) + self.assertEqual("amq-clever-pybook", _r.queue) + self.assertEqual(5, _r.timeout) + _r2: AMQRoute = _f.from_string("amq-clever-pybook:amq-clever-pybook:clever.pybook.localhost.#:5") + self.assertEqual(NULL_ROUTE, _r2) + + def test_validate(self): + _f = AMQRouteFactory() + self.assertTrue(_f.validate("amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0")) + self.assertFalse(_f.validate("amq-clever-pybook:amq-clever-pybook:clever.pybook.localhost.#")) diff --git a/amqp/adapter/test_service_message_factory.py b/amqp/adapter/test_service_message_factory.py new file mode 100644 index 0000000..238ebfc --- /dev/null +++ b/amqp/adapter/test_service_message_factory.py @@ -0,0 +1,46 @@ +from unittest import TestCase + +from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory, RS +from amqp.model.model import ServiceMessageType, CleverMicroServiceMessage + + +class CleverMicroServiceMessageTest(TestCase): + + def test_to_string(self): + _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory('bumble-bee') + message: CleverMicroServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald") + message_str = _f.to_string(message) + assert "|type=01|" in message_str + assert "|body=Duck" in message_str + message: CleverMicroServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald") + message_str = _f.to_string(message) + assert "|type=03|" in message_str + assert "|body=Duck" in message_str + assert "|replyTo=bumble-bee" in message_str + + def test_from(self): + _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router") + message = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald") + serialized = _f.serialize(message) + deserialized = _f.from_bytes(serialized) + assert deserialized.is_valid() + assert deserialized.message_type == ServiceMessageType.ROUTING_DATA + assert deserialized.recipient_name == "Donald" + assert deserialized.message == "Duck" + assert deserialized.reply_to == "amqp-router" + + def test_base64(self): + _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router") + message = _f.as_base64(ServiceMessageType.ROUTING_DATA, "Duck", "Donald") + serialized = _f.serialize(message) + assert b"|81|" in serialized + deserialized = _f.from_bytes(serialized) + assert deserialized.is_valid() + assert deserialized.message_type == ServiceMessageType.ROUTING_DATA + assert deserialized.message == "Duck" + + def test_serialize(self): + _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router") + message = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald") + serialized = _f.serialize(message) + assert RS + "0" + str(ServiceMessageType.ROUTING_DATA.value) diff --git a/amqp/adapter/trace_info_adapter.py b/amqp/adapter/trace_info_adapter.py new file mode 100644 index 0000000..d4650f7 --- /dev/null +++ b/amqp/adapter/trace_info_adapter.py @@ -0,0 +1,14 @@ +from dataclasses import dataclass +from typing import Dict + +@dataclass +class TraceInfoAdapter: + """ + This class should set up the Open Telemetry tracing stack. To be done later. + """ + + @staticmethod + def create_produce_span(message_id: str) -> Dict[str, str]: + return { + "trace-id": message_id + } diff --git a/amqp/config/__init__.py b/amqp/config/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/amqp/config/amq_configuration.py b/amqp/config/amq_configuration.py new file mode 100644 index 0000000..4ea1fd6 --- /dev/null +++ b/amqp/config/amq_configuration.py @@ -0,0 +1,81 @@ +import configparser +import os + + +class AMQAdapter: + + PREFIX: str = 'cm.amq-adapter.' + + def __init__(self, config: configparser.ConfigParser): + """ + cm.amq-adapter.service-name=amq-adapter + cm.amq-adapter.generator-id=164 + cm.amq-adapter.route-mapping= + + :param config: config parser to read configuration values + """ + self.generator_id = config.getint('CleverMicro-AMQ', self.PREFIX + 'generator-id', fallback=164) + self.service_name = config.get('CleverMicro-AMQ', self.PREFIX + 'service-name', fallback='amq-python-adapter-'+self.generator_id) + self.route_mapping = config.get('CleverMicro-AMQ', self.PREFIX + 'route-mapping', fallback=None) + + def adapter_prefix(self) -> str: + return f"{self.service_name}-{self.generator_id}-" + + + +class Dispatch: + + PREFIX: str = 'cm.dispatch.' + + def __init__(self, config: configparser.ConfigParser): + """ + 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.service-host= + cm.dispatch.service-port=8080 + + :param config: config parser to read configuration values + """ + self.use_dlq = config.getboolean('CleverMicro-AMQ', self.PREFIX + 'use-dlq', fallback=False) + self.use_confirms = config.getboolean('CleverMicro-AMQ', self.PREFIX + 'use-confirms', fallback=False) + self.amq_host = config.get('CleverMicro-AMQ', self.PREFIX + 'amq-host', fallback='rabbitmq') + self.amq_port = config.getint('CleverMicro-AMQ', self.PREFIX + 'amq-port', fallback=5672) + self.amq_port_tls = config.getint('CleverMicro-AMQ', self.PREFIX + 'amq-port-tls', fallback=5671) + self.service_host = config.get('CleverMicro-AMQ', self.PREFIX + 'service-host', fallback='rabbitmq') + self.service_port = config.getint('CleverMicro-AMQ', self.PREFIX + 'service-port', fallback=8080) + self.rabbit_mq_user = os.environ["RABBIT_MQ_USER"] + self.rabbit_mq_password = os.environ["RABBIT_MQ_PASSWORD"] + + +class Backpressure: + + PREFIX: str = 'cm.backpressure.' + + def __init__(self, config: configparser.ConfigParser): + """ + 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) + + +class AMQConfiguration: + + def __init__(self): + config = configparser.ConfigParser() + # Read the application.properties file + config.read('config/application.properties') + # + # Add individual config areas + # + self.amq_adapter: AMQAdapter = AMQAdapter(config) + self.dispatch: Dispatch = Dispatch(config) + self.backpressure: Backpressure = Backpressure(config) diff --git a/amqp/config/application.properties b/amqp/config/application.properties new file mode 100644 index 0000000..bda07ba --- /dev/null +++ b/amqp/config/application.properties @@ -0,0 +1,18 @@ +# ==================================================================== +# CleverMicro AMQ Adapter settings +# ==================================================================== +[CleverMicro-AMQ] +cm.amq-adapter.service-name=amq-adapter +cm.amq-adapter.is-router=false +cm.amq-adapter.generator-id=164 +cm.amq-adapter.route-mapping= + +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.backpressure.threshold=20 +cm.backpressure.time-window=3000 +cm.backpressure.management-service-url=management-service:8080 diff --git a/amqp/model/__init__.py b/amqp/model/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/amqp/model/model.py b/amqp/model/model.py new file mode 100644 index 0000000..95eb0f6 --- /dev/null +++ b/amqp/model/model.py @@ -0,0 +1,107 @@ +import base64 + +from typing import List, Mapping, Dict +from dataclasses import dataclass, field +from enum import Enum, auto + +from amqp.model.snowflake_id import SnowflakeId +from amqp.router.utils import sanitize_as_rabbitmq_name + +RS = ":" + + +@dataclass(frozen=True) +class AMQRoute: + """ + Configuration item representing one AMQRoute. Adapter uses this to declare the Exchange + and/or queue and bind the key as routing expression. + + :param componentName: a service/component name that this route points to (owner of the route) + :param exchange: RabbitMQ exchange name + :param queue: RabbitMQ queue name + :param key: RabbitMQ TOPIC Exchange routing key + :param timeout: Timeout in milliseconds, how long to wait for reply, or not waiting if less then 1 + :param validUntil: optional timestamp (millis from epoch) after which the route is ignored + """ + componentName: str + exchange: str + queue: str + key: str + timeout: int + validUntil: int + + FORMAT = "Component-Name:Exchange-Name:Queue-Name:Routing-Key:Timeout:Valid-Until" + + def __str__(self): + return ( + f"{self.componentName}{RS}{self.exchange}{RS}{self.queue}{RS}{self.key}" + f"{RS}{self.timeout}{RS}{self.validUntil}" + ) + + def __eq__(self, other): + if isinstance(other, AMQRoute): + return (self.componentName == other.componentName + and self.queue == other.queue + and self.key == other.key) + return False + + @property + def primary_key(self): + return f"{self.componentName}{RS}{self.queue}{RS}{self.key}" + + def as_string(self) -> str: + return self.__str__() + + +@dataclass +class AMQMessage: + magic: str + id: SnowflakeId + method: str + domain: str + path: str + headers: Mapping[str, List[str]] + trace_info: Mapping[str, str] + base64_body: bytes + + def routing_key(self) -> str: + return sanitize_as_rabbitmq_name(f"{self.domain}.{self.path}") + + def body(self) -> bytes: + return base64.b64decode(self.base64_body) + + +@dataclass +class AMQResponse: + id: str + response_code: int + content_type: str + response: bytes + error: str + error_cause: str + trace_info: Mapping[str, str] + + def body(self) -> bytes: + return base64.b64decode(self.response) + + +class ServiceMessageType(Enum): + ROUTING_DATA_REQUEST = auto() + ROUTING_DATA = auto() + ROUTING_DATA_REMOVE = auto() + ROUTING_DATA_REPLACE = auto() + UNKNOWN = auto() + + +@dataclass(frozen=True) +class CleverMicroServiceMessage: + id: SnowflakeId = None + payload_type: int = 0 + message_type: ServiceMessageType = ServiceMessageType.UNKNOWN + recipient_name: str = None + reply_to: str = None + trace_info: Dict[str, str] = field(default_factory=dict) + message: str = None + + def is_valid(self) -> bool: + return self.message_type is not None diff --git a/amqp/model/snowflake_id.py b/amqp/model/snowflake_id.py new file mode 100644 index 0000000..cdfab3b --- /dev/null +++ b/amqp/model/snowflake_id.py @@ -0,0 +1,68 @@ +import struct + + +class SnowflakeId: + SNOWFLAKE_ID_WIDTH = 10 # with in bytes (8 bits each), can be more than long int width (8 bytes) + + def __init__(self, _input: bytes = None): + self.bits = [0, 0] + if _input is not None: + i = 0 + shift = 8 # starting from Hi value, which is only 2 bytes wide + n = 1 + while i < len(_input) and i <= self.SNOWFLAKE_ID_WIDTH: + self.bits[n] |= _input[i] << shift + if shift == 0: + n -= 1 + shift = 56 + else: + shift -= 8 + i += 1 + + def __init__(self, hi_bits: int, lo_bits: int): + self.bits = [lo_bits, hi_bits] + + @staticmethod + def from_hex(_input): + expected_length = 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH + if _input is None or len(_input) < expected_length: + raise RuntimeError( + f"MessageId is {SnowflakeId.SNOWFLAKE_ID_WIDTH * 8} bits wide, it is represented by HEX string " + f"of length: {expected_length}, real input: [{_input}] of length: {len(_input) if _input else 0}" + ) + + bits = [0, 0] + n = 1 + for i in range(0, expected_length, 2): + if i == expected_length - (2 * 8): + n = 0 + bits[n] = (bits[n] << 8) | int(_input[i:i + 2], 16) + + return SnowflakeId(bits[1], bits[0]) + + def __str__(self): + _hex = f"{self.bits[1]:016X}{self.bits[0]:016X}" + return _hex if len(_hex) <= 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH else _hex[(-2 * SnowflakeId.SNOWFLAKE_ID_WIDTH):] + + def __eq__(self, other): + if isinstance(other, SnowflakeId): + return self.bits[1] == other.bits[1] and self.bits[0] == other.bits[0] + return False + + def __lt__(self, other): + if self.bits[1] < other.bits[1]: + return True + elif self.bits[1] > other.bits[1]: + return False + return self.bits[0] < other.bits[0] + + def get_bytes(self): + high = struct.pack("!Q", self.bits[1]) + low = struct.pack("!Q", self.bits[0]) + return high[2:] + low + + def get_bits(self): + return self.bits + + def as_string(self) -> str: + return self.__str__() diff --git a/amqp/rabbitmq/__init__.py b/amqp/rabbitmq/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/amqp/rabbitmq/rabbit_mq_consumer.py b/amqp/rabbitmq/rabbit_mq_consumer.py new file mode 100644 index 0000000..5fb3097 --- /dev/null +++ b/amqp/rabbitmq/rabbit_mq_consumer.py @@ -0,0 +1,143 @@ +import logging +import pika +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)) + + 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 (pika.connection.exceptions.AMQPError, pika.connection.exceptions.ChannelClosedByBroker) as e: + logging.error(" [E] RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e) + + def register_routes(self): + if self.router.is_open(self.channel): + queue_args = self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None + route_mapping = self.amq_configuration.amq_adapter.route_mapping + logging.info(" [*] RabbitMQConsumer register routes, cfg mapping: [%s]", route_mapping) + requested_routes = self.router.unwrap_route_list(route_mapping) + for route in requested_routes: + queue_name = route.queue() if route.queue() else self.router.get_consume_queue_name() + logging.info(" [*] RabbitMQConsumer register route: [%s]", route) + self.channel.queue_declare(queue_name, durable=True, exclusive=False, + auto_delete=False, arguments=queue_args) + exchange = route.exchange() if route.exchange() else AMQ_RPC_EXCHANGE + self.channel.exchange_declare(exchange, exchange_type=pika.exchange_type.ExchangeType.topic) + self.channel.queue_bind(queue_name, exchange, route.key()) + self.router.register_route( + AMQRoute( + route.component_name(), exchange, 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) + else: + logging.warning(" [W] Channel is null, cannot register routes.") + + def register_deliver_callback(self, callback_provider): + try: + for route in self.router.get_own_routes(): + self.channel.basic_consume( + queue=route.queue(), + on_message_callback=callback_provider(self.channel), + auto_ack=False + ) + logging.info(" [*] Consumer Waiting for messages on: %s", route.queue()) + except pika.connection.exceptions.AMQPError as err: + logging.error(" [E] Callback registration INCOMPLETE, %s", err) + + def register_rpc_handler(self, rpc_consumer): + cm_reply_to_queue = self.router.get_reply_to_queue_name() + self.channel.queue_declare(cm_reply_to_queue, durable=True, exclusive=False, auto_delete=False) + self.channel.exchange_declare( + AMQ_REPLY_TO_EXCHANGE, exchange_type=pika.exchange_type.ExchangeType.direct + ) + self.channel.queue_bind(cm_reply_to_queue, AMQ_REPLY_TO_EXCHANGE, cm_reply_to_queue) + self.channel.basic_consume( + queue=cm_reply_to_queue, + on_message_callback=rpc_consumer, + auto_ack=False + ) + + """ + def limit_rate_inside_time_window(self, message: AMQMessage) -> int: + next_batch_size = BackpressureSubscriber.NO_OP + now = time.time_ns() // 1000000 + queue_size = self.getQueueSize(message) + + # 1. Backpressure? if unprocessed count exceeds threshold, wait for some time + if queue_size > THRESHOLD: + # TODO - signal queue backpressure explicitly to upstream/metrics + # wait: estimated-per-message-processing-time * SQRT(over-the-limit) + self.wait_for_milliseconds( + (WINDOW / self.request_for_window_duration) * (queue_size - THRESHOLD) ** 0.5) + queue_size = self.getQueueSize(message) # refresh queue size after wait + + # 2. batch complete? determine next batch size and wait till window end. reset window. + if self.fulfilled >= self.request_for_window_duration: + next_batch_size = self.calculateNextBatchSize(queue_size) + self.request_for_window_duration = next_batch_size + remaining_time = (self.window_started_at + WINDOW) - now + if remaining_time > 0: + self.wait_for_milliseconds(remaining_time) + queue_size = self.getQueueSize(message) # refresh queue size after wait + self.window_started_at = time.time_ns() // 1000000 + self.fulfilled = 0 + + # 3. Window elapsed? Reset window & recalculate batch size + if now >= self.window_started_at + WINDOW: + if next_batch_size == BackpressureSubscriber.NO_OP: + next_batch_size = self.calculateNextBatchSize(queue_size) + self.request_for_window_duration = next_batch_size + self.window_started_at = time.time_ns() // 1000000 + self.fulfilled = 0 + + # 4. send the message from the current batch + self.send_message(message) + self.fulfilled += 1 + + # 5. report and setup state for the next call + end = time.time_ns() // 1000000 + self.on_next_timestamp = end + return next_batch_size + + def calculateNextBatchSize(self, queue_size: int) -> int: + next_batch_size = self.request_for_window_duration + if queue_size <= THRESHOLD: + # processing proceeds well, can increase the rate + next_batch_size += self.subscriber.increaseRate(self.request_for_window_duration) + logging.debug(f"INCREASE rate to {next_batch_size}. qSize={queue_size}") + else: + decrease = self.subscriber.backoff(queue_size - THRESHOLD) + if self.request_for_window_duration > decrease: + next_batch_size -= decrease + logging.debug(f"DECREASE rate to {next_batch_size}. qSize={queue_size}") + return next_batch_size + """ + """ + def getQueueSize(self, message: AMQMessage) -> int: + try: + route_queue_name = self.router.findRoute(message) + queue_name = route_queue_name.queue if route_queue_name else self.TASK_DISPATCH_QUEUE_NAME + logger.info(f"Get Queue size for {queue_name}") + return max(len(self.awaitingACK), self.channel.message_count(queue_name)) + except Exception: + return len(self.awaitingACK) + """ diff --git a/amqp/rabbitmq/rabbit_mq_producer.py b/amqp/rabbitmq/rabbit_mq_producer.py new file mode 100644 index 0000000..baeab75 --- /dev/null +++ b/amqp/rabbitmq/rabbit_mq_producer.py @@ -0,0 +1,134 @@ +import logging +import time + +import 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 + THRESHOLD = 1000 + WINDOW = 5000 # 5 seconds + + def __init__(self, amq_configuration: AMQConfiguration, router: RouterProducer = None): + self.channel = None + self.connection = None + self.request_for_window_duration = None + self.amq_configuration = amq_configuration + logging.info("*******************************************") + logging.info("******** RabbitMQProducer.init() ********** %s - %s - %s", + self.amq_configuration.dispatch.amq_host, + self.amq_configuration.dispatch.use_confirms, + self.WINDOW) + logging.info("*******************************************") + self.router = router or RouterProducer(amq_configuration) + + # this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); + + self.setup_amq_channel() + self.router.init_routing_paths(self.channel) + + # self.backpressure_per_exchange = defaultdict(lambda: Gauge( + # f"amqp.adapter.exchange.{exchange}", registry=CollectorRegistry())) + + self.router.set_route_added_notifier( + lambda route: logging.info(" [DBG] ---------- RouteAdded, setting up metrics: amqp.adapter.exchange.%s", + route.exchange)) + + self.request_for_window_duration = self.THRESHOLD // 2 + self.window_started_at = time.time() + self.on_next_timestamp = self.window_started_at + self.fulfilled = 0 + + def cleanup(self): + if self.connection: + try: + self.connection.close() + except Exception as e: + logging.error("RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e) + + def setup_amq_channel(self): + """ + Set up the RabbitMQ connection + :return: nothing, it applies changes to instance variables + """ + factory = pika.ConnectionParameters( + host=self.amq_configuration.dispatch.amq_host, + username=self.amq_configuration.dispatch.rabbit_mq_user, + password=self.amq_configuration.dispatch.rabbit_mq_password + ) + self.connection = pika.BlockingConnection(factory) + logging.info("RabbitMQProducer.setupAMQ, got connection, open=%s", self.connection.is_open) + self.connection.add_on_connection_blocked_callback(self.blocked_listener) + self.connection.add_on_connection_unblocked_callback(self.unblocked_listener) + self.channel = self.connection.channel() + self.channel.basic_qos(prefetch_count=self.PREFETCH_COUNT) + if self.amq_configuration.dispatch.use_confirms(): + self.setup_local_ack() + self.channel.add_on_return_callback(self.return_listener) + + 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: + if route.timeout < 0: + # No response expected + self.channel.basic_publish( + route.exchange, + self.router.get_routing_key(message), # context path is a routing key + properties=pika.BasicProperties(delivery_mode=2), + body=AMQMessageFactory.serialize(message) + ) + logging.info(f" [x] basicPublish (no-reply), messageId: {message.id}, to: {route}") + else: + # This is RPC call, there should be response + props = pika.BasicProperties( + correlation_id=message.id, + reply_to=self.router.get_reply_to_queue_name() + ) + self.channel.basic_publish( + route.exchange, + self.router.get_routing_key(message), # context path is a routing key + properties=props, + body=AMQMessageFactory.serialize(message) + ) + 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 {}/{}", message.domain, message.path) + + except Exception as err: + logging.error("Send message failed: %s", err) + + def request_routes(self): + self.router.request_routes_from_adapters() + + def setup_local_ack(self): + self.channel.confirm_delivery() + + def find_route(self, domain: str, path: str) -> AMQRoute: + return self.router.find_route(domain, path) + + def return_listener(self, return_message): + logging.error( + f" [E] ****** (NOT DELIVERED) ID={return_message.properties.message_id}, " + f"Exchg=[{return_message.exchange}], Key=[{return_message.routing_key}], " + f"replyTxt={return_message.reply_text}, length={len(return_message.body) if return_message.body else 0}" + ) + + def blocked_listener(self, reason: str): + logging.error(f" [E] CHANNEL BLOCKED, reason={reason}") + + def unblocked_listener(self): + logging.info(" [*] CHANNEL UNBLOCKED") diff --git a/amqp/router/__init__.py b/amqp/router/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/amqp/router/route_database.py b/amqp/router/route_database.py new file mode 100644 index 0000000..6a17e30 --- /dev/null +++ b/amqp/router/route_database.py @@ -0,0 +1,49 @@ +import logging +import re +from typing import Set + +from amqp.model.model import AMQRoute +from amqp.router.utils import sanitize_as_rabbitmq_name + + +class RouteDatabase: + def __init__(self): + self.routes: Set[AMQRoute] = set() + + def get_routes(self) -> Set[AMQRoute]: + return self.routes + + def pattern(self, routing_key: str) -> re.Pattern: + regex = r"\.".join( + [ + r".*" if token == "" else r"#" if token == "#" else token + for token in routing_key.split(".") + ] + ) + if routing_key.endswith("."): + regex += r"\." + return re.compile(regex) + + def matches(self, routing_key_from_route: str, routing_key_from_message: str) -> bool: + route_pattern = self.pattern(routing_key_from_route) + return bool(route_pattern.match(sanitize_as_rabbitmq_name(routing_key_from_message))) + + def wrap_routes(self) -> str: + return ",".join(str(route) for route in self.routes) + + def find_route(self, domain: str, path: str) -> AMQRoute | None: + routing_key = sanitize_as_rabbitmq_name(f"{domain}.{path}") + logging.info(f" [DBG] findRoute key:{routing_key} in:{self.wrap_routes()}") + for route in self.routes: + if self.matches(route.key, routing_key): + return route + return None + + def add_route(self, route: AMQRoute) -> bool: + if route not in self.routes: + self.routes.add(route) + return True + return False + + def remove_route(self, route: AMQRoute) -> bool: + return route in self.routes and self.routes.remove(route) diff --git a/amqp/router/router_base.py b/amqp/router/router_base.py new file mode 100644 index 0000000..954d4a9 --- /dev/null +++ b/amqp/router/router_base.py @@ -0,0 +1,91 @@ +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.router.route_database import RouteDatabase +from amqp.router.utils import sanitize_as_rabbitmq_name + +CM_REQUEST_EXCHANGE = "CM-Request" +CM_RESPONSE_EXCHANGE = "CM-Response" +CM_C_REQ_QUEUE = "C-REQ" +CM_C_RESP_QUEUE = "C-RESP" +CM_C_AMQ_QUEUE = "C-AMQ" +CM_P_RESP_QUEUE = "P-RESP" +CM_P_REPLY_TO = "P-REPLY-TO" +ANY_RECIPIENT = "*" +AMQ_RPC_EXCHANGE = "AMQ-RPC" +AMQ_REPLY_TO_EXCHANGE = "AMQ-REPLY-TO" +DLQ_EXCHANGE = "DLX" + + +class RouterBase: + def __init__(self): + self.route_database = RouteDatabase() + self.own_routes: Set[AMQRoute] = set() + 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: + return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}") + + def wrap_routes(self) -> str: + return self.route_database.wrap_routes() + + def wrap_own_routes(self) -> str: + return ",".join(str(route) for route in self.own_routes) + + def unwrap_route_list(self, route_string: str) -> List[AMQRoute]: + logging.info(" [DBG] RouterMaster.unwrapRoutes, route(s): %s", route_string) + routes = [AMQRouteFactory.from_string(route_str) for route_str in route_string.split(",")] + return [route for route in routes if route != NULL_ROUTE] + + def get_routes(self) -> Set[AMQRoute]: + return self.route_database.get_routes() + + 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: + return self.route_database.find_route(message.domain, message.path) + + def add_routes(self, message: str) -> None: + try: + amq_route_list = self.unwrap_route_list(message) + logging.info(" [DBG] RouterMaster.addRoute [%d]: %s", len(amq_route_list), message) + for route in amq_route_list: + logging.info(" [*] RouterMaster.addRoute, adding route: %s", route) + if not route.exchange and not route.queue and not route.key: + logging.error("Cannot setup AMQ Exchange for '%s', at least one of " + "Exchange, Queue and/or Key must be present in the routing string. " + "Expected format is '%s'", route, AMQRoute.FORMAT) + else: + if route.exchange != DLQ_EXCHANGE: + is_new_route = self.route_database.add_route(route) + if is_new_route and self.route_added_notifier: + self.route_added_notifier(route) + except Exception as err: + logging.error(" [E] add_routes error: %s", err) + + def set_route_added_notifier(self, callback: Callable[[AMQRoute], None]): + self.route_added_notifier = callback + + def remove_routes(self, message: str) -> None: + try: + to_remove = [AMQRouteFactory.from_string(route_str) for route_str in message.split(",")] + removed_count = sum(1 for route in to_remove if self.remove_route(route)) + print(f" [*] RouterMaster.removeRoutes(): {message}, removed={removed_count}") + except IOError as err: + print(f" [E] Can't remove route(s): {err}") + + def remove_route(self, route: AMQRoute) -> bool: + is_route_removed = self.route_database.remove_route(route) + if is_route_removed and self.route_removed_notifier: + self.route_removed_notifier(route) + return is_route_removed + + def add_route(self, route: AMQRoute) -> None: + self.route_database.add_route(route) + + def add_own_route(self, route: AMQRoute) -> None: + self.own_routes.add(route) diff --git a/amqp/router/router_consumer.py b/amqp/router/router_consumer.py new file mode 100644 index 0000000..cd10e9a --- /dev/null +++ b/amqp/router/router_consumer.py @@ -0,0 +1,143 @@ +import logging +from typing import Set + +from pika import spec +from pika.adapters.blocking_connection import BlockingChannel + +from amqp.adapter.amq_route_factory import AMQRouteFactory +from amqp.config.amq_configuration import AMQConfiguration +from amqp.model.model import CleverMicroServiceMessage, ServiceMessageType, AMQRoute +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 + + +class RouterConsumer(RouterProducer): + """ + * Contains Consumer logic. The logic here: + * Sets a listener on CM_C_REQ_QUEUE queue to receive routing data requests + * Upon receipt of routing data request, responds on CM_RESPONSE_EXCHANGE with local routing data. + * It also proactively pushes the local routing data to CM_RESPONSE_EXCHANGE on startup to + * notify the other (excluding self) Producers about its presence. + * + * Consumer is also Producer (Consumer sits alongside the CM Service, and anytime the Service needs + * to communicate with other Service, it can do so via the Adapter). Therefore, the Producer + * capability is also required. + """ + + def __init__(self, configuration: AMQConfiguration): + super().__init__(configuration) + self.amqRouteFactory = AMQRouteFactory() + logging.info(" [*] RouterConsumer.") + + def init_routing_paths_consumer(self, _channel: BlockingChannel): + """ + * Initialize router according to Consumer mode. + """ + + self.init_routing_paths(_channel) + logging.info(" [*] RouterConsumer.initConsumerRouter, channel open::", self.is_open(_channel)) + if self.is_open(_channel): + try: + # 1. declare a Queue that gives the RoutingDAta requests + cm_request_queue: str = self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_REQ_QUEUE + self.channel.queue_declare(cm_request_queue, durable=True, exclusive=False, auto_delete=False) + # 1a. bind queue to CM_RESPONSE_EXCHANGE w/o any conditions (empty routing key) + self.channel.queue_bind(queue=cm_request_queue, exchange=CM_REQUEST_EXCHANGE, routing_key="") + # 1b. Service adapter mode - listens on Request queue and replies to Response queue + logging.info(" [*] RouterConsumer listens for RouteDataRequests on:", cm_request_queue) + self.channel.basic_consume(queue=cm_request_queue, + auto_ack=False, + on_message_callback=self.handle_routing_data_request_message) + except Exception as ioe: + logging.error(" [E] RouterConsumer FATAL: cannot initialize, %s", ioe) + else: + logging.error(" [E] RouterConsumer FATAL: cannot initialize, RabbitMQ channel is not open.") + + """ + * ServiceMessageType.ROUTING_DATA_REQUEST handler listening on queue 'CleverMicroRequest' + """ + + def handle_routing_data_request_message(self, + ch: BlockingChannel, + method: spec.Basic.Deliver, + properties: spec.BasicProperties, + body: bytes): + # some debug statements, TODO - remove for prod release + delivery_tag = method.delivery_tag + debug = "Delivery.Properties = " + properties.__str__() + logging.info(" [DBG] ******[%] (ROUTING_DATA_REQ) tag: %s, env: %s, props: %s", + delivery_tag, method.consumer_tag, method.get_properties(), debug) + + self.channel.basic_ack(delivery_tag, False) + + message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(body) + route_mapping: str = self.wrap_own_routes() + logging.info(" [DBG] ******[%s] (ROUTING_DATA_REQ) msg=%, routeMap=[{]", + delivery_tag, self.service_message_factory.to_string(message), route_mapping) + # Reply only if self adapter has own routeMapping (e.g. API-gateway-side adapter has none) + if message.is_valid() and message.message_type == ServiceMessageType.ROUTING_DATA_REQUEST and route_mapping: + try: + service_message: CleverMicroServiceMessage = self.service_message_factory.of( + ServiceMessageType.ROUTING_DATA, route_mapping, message.reply_to + ) + self.publish_service_message(service_message, CM_RESPONSE_EXCHANGE) + except Exception as err: + logging.error(" [E] ******[%s] (ROUTING_DATA_REQ) Response to queue: %s failed: %s", + delivery_tag, CM_RESPONSE_EXCHANGE, err) + + else: + logging.error(" [E] ******[%s] (ROUTING_DATA_REQ) [%s] produces invalid or no CM msg type.", + delivery_tag, str(body)) + + def register_route(self, route: AMQRoute): + """ + * Called in Consumer (Service Adapter) context. Registers the routing data for self Service + * with the Master router by sending own routing data to Master via CleverMicroResponse queue. + * :param route: route to register with Master + """ + try: + if self.is_open(self.channel) and route.queue: + serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of( + ServiceMessageType.ROUTING_DATA, route.as_string(), ANY_RECIPIENT + ) + self.publish_service_message(serviceMessage, CM_RESPONSE_EXCHANGE) + self.add_own_route(route) + logging.info(" [DBG] RouterConsumer registered Route::", route) + else: + logging.warning(" [W] RouterConsumer couldn't register route data: %s channel isn't open.", + CM_RESPONSE_EXCHANGE) + except Exception as ioe: + logging.error(" [E] RouterConsumer failed register route data: %s", ioe) + + def remove_route_on_cleanup(self, route: AMQRoute) -> bool: + """ + * Unregister / remove route from the list + * :param route: route to remove + * :return result of remove operation + """ + logging.info(" [DBG] RouterConsumer.removeRoute, route::", route.as_string()) + try: + service_message: CleverMicroServiceMessage = self.service_message_factory.of( + ServiceMessageType.ROUTING_DATA_REMOVE, + route.as_string(), + ANY_RECIPIENT + ) + if not self.publish_service_message(service_message, CM_RESPONSE_EXCHANGE): + logging.warning(" [W] RouterConsumer couldn't remove current route: %s, channel not open.", + route) + + except Exception as ioe: + logging.error(" [E] RouterConsumer failed remove current route: %s", ioe) + + result: bool = self.remove_route(route) + return result + + def cleanup(self): + """ + * Cleanup - de-register routes with master. + """ + routes: Set[AMQRoute] = self.get_routes() + logging.info(" [DBG] RouterConsumer.cleanup() [%d]: %s", len(routes), self.wrap_routes()) + not_removed = sum(1 for r in (self.remove_route_on_cleanup(route) for route in routes) if not r) + if not_removed > 0: + logging.info(" [W] %d route(s) not removed. self may lead to failed delivery on self route", not_removed) diff --git a/amqp/router/router_producer.py b/amqp/router/router_producer.py new file mode 100644 index 0000000..4244358 --- /dev/null +++ b/amqp/router/router_producer.py @@ -0,0 +1,190 @@ +""" + * Maps the request to RabbitMQ exchange. Handles the service discovery logic. + * + * Router can be either in 'Master' or 'Service (Consumer)' modes. 'Master' mode is the outward + * facing side that receives the requests and routes it to queue associated with requested service. + * 'Service' mode is consumer side for given service, which receives the message from AMQ and + * forwards it to the application. + """ +import logging +import os +import pika +from pika import BasicProperties, spec +from pika.adapters.blocking_connection import BlockingChannel +from pika.exchange_type import ExchangeType + +from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory +from amqp.config.amq_configuration import AMQConfiguration +from amqp.model.model import CleverMicroServiceMessage, 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 + + +class RouterProducer(RouterBase): + + def __init__(self, configuration: AMQConfiguration): + """ + * Initialize router - need to supply the config values and RabbitMQ channel. + * :param configuration: adapter's configuration reference. + """ + super().__init__() + self.connection = None + self.service_message_factory = CleverMicroServiceMessageFactory(configuration.amq_adapter.service_name) + self.amq_configuration: AMQConfiguration = configuration + self.channel: pika.adapters.blocking_connection.BlockingChannel | None = None + + logging.info(f" [*] RouterProducer.: %s, route: %s", + self.amq_configuration.amq_adapter.service_name, + self.amq_configuration.amq_adapter.route_mapping) + + def amq_consumer(self): + credentials = pika.PlainCredentials( + os.environ['RABBIT_MQ_USER'], os.environ['RABBIT_MQ_PASSWORD'] + ) + self.connection = pika.BlockingConnection( + pika.ConnectionParameters(host='localhost', credentials=credentials) + ) + self.channel = self.connection.channel() + self.channel.start_consuming() + + def init_routing_paths(self, channel: pika.adapters.blocking_connection.BlockingChannel | None): + """ + * Initialize router according to Producer mode. + * 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 + * on CleverMicroResponse for any service responding with its routing data. + """ + + self.channel = channel + logging.info(" [*] RouterProducer.initRoutingPaths, channel open: %s", self.is_open(self.channel)) + if self.is_open(self.channel): + try: + # **** set up the service discovery internal exchanges **** + # 1. declare RequestExchange to where to publish the RoutingData requests + self.channel.exchange_declare(exchange=CM_REQUEST_EXCHANGE, exchange_type=ExchangeType.fanout) + # + # 2. declare ResponseExchange where to publish current RoutingData (response to + # a RoutingData request) + self.channel.exchange_declare(exchange=CM_RESPONSE_EXCHANGE, exchange_type=ExchangeType.fanout) + # 2a. declare a Response queue for Routing Data replies from Services + cm_response_queue: str = self.get_response_queue_name() + self.channel.queue_declare(cm_response_queue, durable=True, exclusive=False, auto_delete=False, + arguments={}) + # 2b. bind queue to CM_RESPONSE_EXCHANGE w/o any conditions (empty routing key) + self.channel.queue_bind(queue=cm_response_queue, exchange=CM_RESPONSE_EXCHANGE, routing_key="#") + # 2c. listen for incoming RoutingData messages + self.channel.basic_consume( + queue=cm_response_queue, + auto_ack=False, + on_message_callback=self.handle_routing_data_message) + self.channel.start_consuming() + logging.info(" [*] Routing master listens for Route updates on %s", cm_response_queue) + except Exception as ioe: + logging.error(" [E] RouterProducer FATAL: cannot initialize, %s", ioe) + + else: + logging.error(" [E] RouterProducer FATAL: cannot initialize, RabbitMQ channel is not open.") + + def handle_routing_data_message(self, + ch: BlockingChannel, + method: spec.Basic.Deliver, + properties: spec.BasicProperties, + body: bytes): + """ + * Handler for CLEVER_MICRO_RESPONSE queue messages. Used in Master mode to handle receipt of + * routing data, invokes addRoute or removeRoute depending on the type of the message. + """ + print(f" [x] Received {body}, m={method}, p={properties}, ch={ch}") + """ + logging.info(" [DBG] ******[%s] ROUTING_DATA %s >>> %s", + delivery.getEnvelope().getDeliveryTag(), + delivery.getEnvelope().toString(), str(delivery.getBody())) + """ + # ACK message now so that it is not being redelivered + self.channel.basic_ack(method.delivery_tag, False) + message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(body) + # ignore the message if it comes from ourselves (even if from different instance) + logging.info(" [DBG] ******[%s] ROUTING_DATA, msg=%s, routeMap=[{] sn={, replyTo={", + method.delivery_tag, + self.service_message_factory.to_string(message), + self.amq_configuration.amq_adapter.route_mapping, + self.amq_configuration.amq_adapter.service_name, + message.reply_to) + if message.is_valid() and not self.amq_configuration.amq_adapter.service_name == message.reply_to: + if message.message_type == ServiceMessageType.ROUTING_DATA: + self.add_routes(message.message) + if message.message_type == ServiceMessageType.ROUTING_DATA_REMOVE: + self.remove_routes(message.message) + + def request_routes_from_adapters(self): + """ + * Send a broadcast message prompting all listening adapters to reply with own routing data. + """ + logging.info(" [*] RouterProducer.requestRoutesFromAdapters, queue name: %s", CM_REQUEST_EXCHANGE) + try: + serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of( + ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT + ) + if not self.publish_service_message(serviceMessage, CM_REQUEST_EXCHANGE): + logging.warning(" [E] RouterProducer could not request route data: %s channel is not open.", + CM_REQUEST_EXCHANGE) + + except Exception as ioe: + logging.error(" [E] RouterProducer failed request routing data: %s", ioe) + + BASIC: BasicProperties = BasicProperties(content_type="application/octet-stream", + content_encoding=None, + headers=None, + delivery_mode=1, + priority=0, + correlation_id=None) + + def publish_service_message(self, message: CleverMicroServiceMessage, exchange_name: str) -> bool: + """ + * Publishes a service message to the service queue and logs success log entry. + * + * :param message: Service Message to be sent + * :param exchangeName: target exchange name where to publish the message + * :param queueName target: queue name (Optional, purpose is to get the queue size only) + * :return True if channel is open, False otherwise + * @throws IOException if something else goes wrong + """ + if self.is_open(self.channel): + binary_content: bytes = self.service_message_factory.serialize(message) + self.channel.basic_publish(exchange=exchange_name, + routing_key="", # empty routing key + properties=self.BASIC, + body=binary_content) + logging.info(" [x] Service Message Sent to %s, msg: %s", exchange_name, str(binary_content)) + return True + return False + + def get_reply_to_queue_name(self) -> str: + """ + * self should create a unique name for self process and also when there's several instances. + * @return unique reply-to queue name. + """ + return self.amq_configuration.amq_adapter.adapter_prefix() + CM_P_REPLY_TO + + def get_response_queue_name(self) -> str: + """ + * self should create a unique name for self process and also when there's several instances. + * @return unique reply-to queue name. + """ + return self.amq_configuration.amq_adapter.adapter_prefix() + CM_P_RESP_QUEUE + + def get_consume_queue_name(self) -> str: + """ + * self should create a unique name for self process and also when there's several instances. + * @return unique reply-to queue name. + """ + return self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_AMQ_QUEUE + + def is_open(self, _channel: pika.adapters.blocking_connection.BlockingChannel) -> bool: + """ + * Test if the channel is usable (opened). + * :param _channel: the channel + * :return True if the channel is opened. + """ + return _channel is not None and _channel.is_open diff --git a/amqp/router/serializer.py b/amqp/router/serializer.py new file mode 100644 index 0000000..8a916c7 --- /dev/null +++ b/amqp/router/serializer.py @@ -0,0 +1,61 @@ +import struct +from io import BytesIO +from typing import List, Union, Dict + + +def long_to_bytes(x: int) -> bytes: + return struct.pack('q', x) + + +def bytes_to_long(b: bytes) -> int: + return struct.unpack('q', b)[0] + + +def read_long(bis: BytesIO) -> int: + try: + return struct.unpack('q', bis.read(8))[0] + except IOError: + return 0 + + +def object_or_list_as_string(value: Union[str, List[str]]) -> str: + if isinstance(value, list): + return f"[{','.join(value)}]" + return str(value) + + +def map_as_string(map_: Dict[str, str]) -> str: + return ','.join(f"{k}={object_or_list_as_string(v)}" for k, v in map_.items()) + + +def map_with_list_as_string(map_: Dict[str, List[str]]) -> str: + return ','.join(f"{k}={object_or_list_as_string(v)}" for k, v in map_.items()) + + +def parse_map_string(input_str: str) -> Dict[str, str]: + map_ = {} + if input_str.startswith('{') and input_str.endswith('}'): + for entry in input_str[1:-1].split(','): + if len(entry) > 1: + key, value = entry.split('=') + map_[key] = value + return map_ + + +def csv_as_list(input_str: str) -> List[str]: + return [item.strip() for item in input_str.strip('[]').split(',')] + + +def add_to_map(map_: Dict[str, List[str]], token: str) -> None: + eq_idx = token.index('=') + if eq_idx > 0: + key = token[:eq_idx] + map_[key] = csv_as_list(token[eq_idx+1:]) + + +def parse_map_list_string(input_str: str) -> Dict[str, List[str]]: + map_ = {} + if input_str.startswith('{') and input_str.endswith('}'): + for token in input_str[1:-1].split('],'): + add_to_map(map_, token) + return map_ diff --git a/amqp/router/test_route_database.py b/amqp/router/test_route_database.py new file mode 100644 index 0000000..76af304 --- /dev/null +++ b/amqp/router/test_route_database.py @@ -0,0 +1,24 @@ +from unittest import TestCase + + +class TestRouteDatabase(TestCase): + def test_get_routes(self): + self.fail() + + def test_pattern(self): + self.fail() + + def test_matches(self): + self.fail() + + def test_wrap_routes(self): + self.fail() + + def test_find_route(self): + self.fail() + + def test_add_route(self): + self.fail() + + def test_remove_route(self): + self.fail() diff --git a/amqp/router/test_router_base.py b/amqp/router/test_router_base.py new file mode 100644 index 0000000..46fd494 --- /dev/null +++ b/amqp/router/test_router_base.py @@ -0,0 +1,39 @@ +from unittest import TestCase + + +class TestRouterBase(TestCase): + def test_get_routing_key(self): + self.fail() + + def test_wrap_routes(self): + self.fail() + + def test_wrap_own_routes(self): + self.fail() + + def test_unwrap_route_list(self): + self.fail() + + def test_get_routes(self): + self.fail() + + def test_find_route(self): + self.fail() + + def test_find_route_by_message(self): + self.fail() + + def test_add_routes(self): + self.fail() + + def test_remove_routes(self): + self.fail() + + def test_remove_route(self): + self.fail() + + def test_add_route(self): + self.fail() + + def test_add_own_route(self): + self.fail() diff --git a/amqp/router/test_utils.py b/amqp/router/test_utils.py new file mode 100644 index 0000000..cc96abe --- /dev/null +++ b/amqp/router/test_utils.py @@ -0,0 +1,15 @@ +from unittest import TestCase + +from amqp.router.utils import sanitize_as_rabbitmq_name + + +class Test(TestCase): + def test_sanitize_as_rabbitmq_name(self): + self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa.bb/cc")) + self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa-bb/cc")) + self.assertEqual("aa", sanitize_as_rabbitmq_name("aa&bb/cc")) + self.assertEqual("aa", sanitize_as_rabbitmq_name("aa+bb/cc")) + self.assertEqual("aa", sanitize_as_rabbitmq_name("aa%bb/cc")) + self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa#bb/cc")) + self.assertEqual("aa", sanitize_as_rabbitmq_name("aa?bb////cc")) + self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa///bb////cc")) diff --git a/amqp/router/utils.py b/amqp/router/utils.py new file mode 100644 index 0000000..8ffe5d5 --- /dev/null +++ b/amqp/router/utils.py @@ -0,0 +1,9 @@ +import re + + +def sanitize_as_rabbitmq_name(input_str: str) -> str: + pattern = re.compile(r"[^a-zA-Z0-9.#/\-]") + matcher = pattern.search(input_str) + token = input_str[:matcher.start()] if matcher else input_str + step1 = re.sub(r"[^A-Za-z0-9]", ".", token) + return re.sub(r"\.+", ".", step1) diff --git a/amqp/service/__init__.py b/amqp/service/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py new file mode 100644 index 0000000..c3faf71 --- /dev/null +++ b/amqp/service/amq_service.py @@ -0,0 +1,84 @@ +import os +import logging +from functools import partial +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import ( + ConsoleSpanExporter, + SimpleSpanProcessor +) +from opentelemetry.propagate import set_global_textmap +from opentelemetry.propagators.tracecontext import TraceContextTextMapPropagator + +from amqp.adapter.amq_message_factory import AMQMessageFactory +from amqp.adapter.amq_route_factory import AMQRouteFactory +from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory +from amqp.rabbitmq.rabbit_mq_consumer import RabbitMQConsumer +from amqp.service.service_message_handler import AMQServiceMessageHandler + + +class AMQService: + CLEVER_AMQP_CLIENT_VERSION = os.getenv("CLEVER_AMQP_CLIENT_VERSION") + MAX_WAIT = 300 # defaults to 5 minutes (300 seconds) + + def __init__(self, amq_configuration, http_adapter): + self.amq_configuration = amq_configuration + self.http_adapter = http_adapter + self.message_factory = AMQMessageFactory.get_instance(self.amq_configuration.amq_adapter().generator_id()) + self.rabbit_mq_consumer = RabbitMQConsumer(self.amq_configuration) + self.amq_service_message_handler = AMQServiceMessageHandler(self.http_adapter, self.get_tracer()) + + def init(self): + CleverMicroServiceMessageFactory.process_name(self.amq_configuration.amq_adapter().service_name()) + logging.info("********* AMQService.isRouter = %s ***********", self.amq_configuration.amq_adapter().is_router()) + logging.info("***********************************************") + logging.info("********** AMQ Service *******************") + logging.info("***********************************************") + + self.setup_opentelemetry() + self.register_routes() + self.register_rpc_handler() + self.rabbit_mq_consumer.request_routes() + + def cleanup(self): + logging.info(" [DBG] AMQService.cleanup() rmqc=%s", self.rabbit_mq_consumer) + if self.rabbit_mq_consumer: + self.rabbit_mq_consumer.cleanup() + + def reinitialize_if_disconnected(self): + if not self.rabbit_mq_consumer.channel or not self.rabbit_mq_consumer.channel.is_open: + self.init() + + def setup_opentelemetry(self): + trace.set_tracer_provider(TracerProvider()) + trace.get_tracer_provider().add_span_processor( + SimpleSpanProcessor(ConsoleSpanExporter()) + ) + set_global_textmap(TraceContextTextMapPropagator()) + self.tracer = trace.get_tracer("com.cleverthis") + logging.info("OTEL %s %s %s", trace.get_logger_bridge(), trace.get_meter_provider(), trace.get_tracer_provider()) + logging.info("OTEL TRACER %s", trace.get_propagators()) + + def register_routes(self): + route_mapping = self.amq_configuration.amq_adapter().route_mapping() + if AMQRouteFactory.validate(route_mapping): + try: + self.rabbit_mq_consumer.register_routes() + self.rabbit_mq_consumer.register_deliver_callback( + partial(self.amq_service_message_handler.on_delivery_callback_http, self.rabbit_mq_consumer.channel) + ) + self.rabbit_mq_consumer.register_rpc_handler( + self.amq_service_message_handler.on_reply_consumer(self.rabbit_mq_consumer.channel) + ) + except Exception as e: + raise RuntimeError(e) + elif route_mapping: + logging.info(" [W] route [%s] failed validation, no routes were registered.", route_mapping) + + def register_rpc_handler(self): + try: + self.rabbit_mq_consumer.register_rpc_handler( + self.amq_service_message_handler.on_reply_consumer(self.rabbit_mq_consumer.get_channel()) + ) + except Exception as err: + raise diff --git a/amqp/service/service_message_handler.py b/amqp/service/service_message_handler.py new file mode 100644 index 0000000..79964d7 --- /dev/null +++ b/amqp/service/service_message_handler.py @@ -0,0 +1,143 @@ +import logging +import time + +from typing import Callable, Dict +from concurrent.futures import Future +from opentelemetry import trace +from opentelemetry.propagate import get_global_textmap, set_global_textmap +from opentelemetry.propagators.tracecontext import TraceContextTextMapPropagator +from opentelemetry.trace import Span, Tracer, TraceState + +from amqp.adapter.amq_message_factory import AMQMessageFactory +from amqp.model.model import AMQResponse +from amqp.router.serializer import map_as_string + + +class AMQServiceMessageHandler: + def __init__(self, http_adapter, tracer): + self.http_adapter = http_adapter + self.tracer = tracer + self.outstanding = {} + + def on_delivery_callback_http(self, channel, consumerTag, delivery): + start = time.time() + deliveryTag = delivery.envelope.delivery_tag + debug = f" [DBG] ######[{deliveryTag}] (AMQ MESSAGE), env: {delivery.envelope}, props: {delivery.properties}" + logging.info(debug) + + amq_message = AMQMessageFactory.from_bytes(delivery.body) + logging.info(" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, traceInfo=%s", + deliveryTag, consumerTag, amq_message.id, map_as_string(amq_message.trace_info)) + + propagator = TraceContextTextMapPropagator.get_instance() + context = propagator.extract(trace.get_current_context(), amq_message.trace_info , self.get_text_getter()) + parent_span = Span.from_context(context) + context = context.with_span(parent_span) + + with context.activate(): + 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 = self.http_adapter.handle(amq_message) + future.add_done_callback( + lambda f: self.on_http_response(delivery, deliveryTag, channel, f.result()) + ) + else: + logging.error(" [E] Unknown MAGIC value: [%s], don't know how to interpret the message", + amq_message.magic()) + + duration = time.time() - start + if "___NACK___" in str(amq_message.body()): + logging.info(" [x] ######[%s] Done, single NACK: %sms.", deliveryTag, duration) + channel.basic_nack(deliveryTag, False, False) + else: + logging.info(" [x] ######[%s] Done, single ACK: %sms.", deliveryTag, duration) + channel.basic_ack(deliveryTag, False) + + def on_http_response(self, delivery, deliveryTag, channel, http_response): + response = AMQResponse( + delivery.properties.correlation_id, + http_response.status_code, + http_response.headers.get("Content-Type"), + http_response.body, + http_response.status_message, + "", + self.collect_trace_info(trace.get_current_context()) + ) + try: + self.send_reply(delivery.properties.reply_to, deliveryTag, channel, response) + except Exception as e: + raise RuntimeError(e) + + + class AMQServiceMessageHandler: + def __init__(self, http_adapter, tracer): + self.http_adapter = http_adapter + self.tracer = tracer + self.outstanding: Dict[str, 'ResponseHolder'] = {} + + def collect_trace_info(self, context): + map = {} + TraceContextTextMapPropagator.get_instance().inject( + context if context else trace.get_current_context(), + map, self.set_text + ) + return map + + def send_reply(self, reply_to, delivery_tag, channel, response): + if reply_to: + logging.info(" [*] ######[%s] AMQ Message Reply, To=%s, msgId=%s", + delivery_tag, reply_to, response.id()) + reply_props = pika.BasicProperties( + correlation_id=response.id(), + content_type=response.content_type() + ) + channel.basic_publish( + RouterBase.AMQ_REPLY_TO_EXCHANGE, + reply_to, + reply_props, + AMQResponseFactory.serialize(response) + ) + + def on_reply_consumer(self, channel): + def handle_delivery(consumerTag, envelope, properties, body): + reply_id = properties.correlation_id + content_type = properties.content_type + delivery_tag = envelope.delivery_tag + debug = f" [DBG] ######[{delivery_tag}] (RPC REPLY) env: {envelope}, props: {properties}, BODY: {body.decode()}" + logging.info(debug) + print(f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}") + + with self.outstanding.lock: + reply = 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() + reply.response().set_result(Response.ok(result.body(), content_type).build()) + + channel.basic_ack(delivery_tag, False) + + return handle_delivery + + def register_response_holder(self, id, response_holder): + self.outstanding[id] = response_holder + logging.info(" [*] AMQMessage Sent, OUTSTANDING response holder registered: %s, size=%s", + self.outstanding, len(self.outstanding)) + + def get_default_trace_state(self): + return TraceState(entries={}) + + def get_text_getter(self): + return lambda map, key: map.get(key) + + def set_text(self, carrier, key, value): + carrier[key] = value diff --git a/amqp/service/tracing.py b/amqp/service/tracing.py new file mode 100644 index 0000000..03db010 --- /dev/null +++ b/amqp/service/tracing.py @@ -0,0 +1,89 @@ +from opentelemetry import metrics, trace +from opentelemetry.instrumentation.flask import FlaskInstrumentor +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter + +from prometheus_client import start_http_server +from opentelemetry.exporter.prometheus import PrometheusMetricReader +from opentelemetry.sdk.resources import SERVICE_NAME, Resource + +import os + +def initialize_trace(flask_app, name): + # Resource can be required for some backends, e.g. Jaeger or Prometheus + # If resource wouldn't be set - traces wouldn't appear in Jaeger. + # The 'name' value is the name shown in Jaeger. + resource = Resource(attributes={ + SERVICE_NAME: name if name else os.environ.get('APPLICATION_NAME', 'Python App') + }) + # + # 1. Tracing (Jaeger) configuration + # ---------------------------------- + # + # Initialize the tracer provider. This is abstract API call. + # The actual implementation is given by the included dependency + # The opentelemetry default is the OTLP trace provider. + trace.set_tracer_provider(TracerProvider(resource=resource)) + # + # Configure the OTLP exporter (or HTTP exporter the same way) + # the recommended configuration is via environment variables. The mandatory ones: + # OTEL_EXPORTER_OTLP_ENDPOINT: "http://jaeger:4317" + # OTEL_EXPORTER_OTLP_INSECURE: "true" + # for more see here: https://opentelemetry-python.readthedocs.io/en/latest/exporter/otlp/otlp.html + # alternatively these values can be provided as input values to OTLPSpanExporter() constructor. + # + otlp_exporter = OTLPSpanExporter() + # + # Add the exporters to the batch span processor + span_processor = BatchSpanProcessor(otlp_exporter) + trace.get_tracer_provider().add_span_processor(span_processor) + + print(' * Instrumenting Flask application..... ') + FlaskInstrumentor().instrument_app(flask_app) + # + # Python Flask is now automatically instrumented to export details of each REST endpoint call. + # For Django based app, use 'opentelemetry-instrumentation-django' import instead. see here: + # https://opentelemetry-python.readthedocs.io/en/latest/examples/django/README.html for example + # + # If the used library has no instrumentation support in opentelemetry.io, then the trace + # can be started manually. In this code, + # 'trace' should be global shared variable, initialized once per application (see tracing.py) + # 'tracer' can be shared global or local variable, application can use multiple uniquely + # named tracers, Jaeger will group the generated traces by the name of the tracer + # that was used to start the Span. See this example code: + # + # tracer = trace.get_tracer(name) + # + # @app.route("/server_request") + # def server_request(): + # with tracer.start_as_current_span( + # "server_request", + # context=extract(request.headers), + # kind=trace.SpanKind.SERVER, + # attributes=collect_request_attributes(request.environ), + # ): + # # do the work here, like print something and return response + # print(request.args.get("param")) + # return "served" + # + # Note: + # the otlp code looks for HTTP header named 'traceparent' that, if present, should contain + # the otlp parent span identification, and the span created here will be made a child + # of that parent span, indicating visually the progression of requests via different services + # for as long as the current span is provided in the traceparent header to next service. + + + # 2. Metrics (Prometheus) configuration + # ------------------------------------- + # + # Start Prometheus client + start_http_server(port=9464, addr="localhost") + # Initialize PrometheusMetricReader which pulls metrics from the SDK + # on-demand to respond to scrape requests (unlike Jaeger, which worls with push model - application + # pushes traces when those become available to Jaeger's collector. Prometheus works with pull model + # where it periodically polls pre-configured and/or discovered endpoints for metrics. + reader = PrometheusMetricReader() + provider = MeterProvider(resource=resource, metric_readers=[reader]) + metrics.set_meter_provider(provider) diff --git a/logging.conf b/logging.conf new file mode 100644 index 0000000..2d64dd2 --- /dev/null +++ b/logging.conf @@ -0,0 +1,34 @@ +[loggers] +keys=root,myapp + +[handlers] +keys=consoleHandler,fileHandler + +[formatters] +keys=simpleFormatter + +[logger_root] +level=DEBUG +handlers=consoleHandler + +[logger_myapp] +level=DEBUG +handlers=consoleHandler,fileHandler +qualname=myapp +propagate=0 + +[handler_consoleHandler] +class=StreamHandler +level=DEBUG +formatter=simpleFormatter +args=(sys.stdout,) + +[handler_fileHandler] +class=FileHandler +level=DEBUG +formatter=simpleFormatter +args=('AMQ_PyAdapter.log',) + +[formatter_simpleFormatter] +format=%(asctime)s - %(name)s - %(levelname)s - %(message)s +datefmt=%Y-%m-%d %H:%M:%S diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9049629 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,8 @@ +aiohttp~=3.11.11 +pika~=1.3.2 +flask +opentelemetry-api +opentelemetry-sdk +opentelemetry-exporter-otlp +opentelemetry-exporter-prometheus +opentelemetry-instrumentation-flask