From 41fbe46129f4a813624553d91ab5450dddec2d34 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Sat, 26 Jul 2025 15:29:07 +0100 Subject: [PATCH] feat/#63 - cleverswarm unidirectional command --- amqp/adapter/backpressure_handler.py | 43 ++++++++++++---------- amqp/adapter/data_message_factory.py | 37 ++++++++++--------- amqp/adapter/serializer.py | 10 ++++-- amqp/router/router_base.py | 7 +++- amqp/service/amq_message_handler.py | 33 ++++++++++------- amqp/service/amq_service.py | 6 ++-- cleverbrag/clever_brag_adapter.py | 4 +-- cleverswarm/clever_swarm_adapter.py | 4 +-- demo.py | 53 +++++++++++++++++----------- otdemo/otdemo.properties | 8 ++--- otdemo/otdemo_adapter.py | 4 +-- otdemo/otdemo_backend.py | 25 +++++++++++++ 12 files changed, 148 insertions(+), 86 deletions(-) create mode 100644 otdemo/otdemo_backend.py diff --git a/amqp/adapter/backpressure_handler.py b/amqp/adapter/backpressure_handler.py index 478387f..e475a30 100644 --- a/amqp/adapter/backpressure_handler.py +++ b/amqp/adapter/backpressure_handler.py @@ -78,6 +78,7 @@ class BackpressureHandler: self.current_availability = 1 self.max_load = 1 self.current_load = 0 + self.idle_state_detected_time = 0 def increase_current_load(self): """Increase the number of parallel executions, or current_availability load""" @@ -146,12 +147,12 @@ class BackpressureHandler: current_time = time.time() last_event_delta: float = current_time - self.last_backpressure_event_time - logging_info( - "Backpressure: Check conditions, current_availability=%s, max=%s, last_activity=%s", - self.current_availability, - self.max_availability, - last_event_delta, - ) + # logging_info( + # "Backpressure: Check conditions, current_availability=%s, max=%s, last_activity=%s", + # self.current_availability, + # self.max_availability, + # last_event_delta, + # ) # Check for OVERLOAD condition - low availability (less than 20% of maximum) if self.current_availability <= round(0.1 * self.max_availability): @@ -171,7 +172,7 @@ class BackpressureHandler: idle_duration = self.config.backpressure.idle_duration # Check if service has been idle for longer than the configured duration if ( - last_event_delta > idle_duration + current_time - self.idle_state_detected_time > idle_duration and self.last_backpressure_event == ScalingRequestAlert.IDLE ): logging_info( @@ -184,19 +185,25 @@ class BackpressureHandler: await self._handle_backpressure_idle_event() # update / reset time-window so that the IDLE is not sent too often self.last_backpressure_event_time = current_time + self.idle_state_detected_time = current_time + # send IDLE even only once + self.last_backpressure_event = ScalingRequestAlert.UPDATE else: # Don't send IDLE right away when the availability increases, but wait for a while, send UPDATE instead self.last_backpressure_event = ScalingRequestAlert.IDLE - _scaling_request: ScaleRequestV1 = ScaleRequestV1( - self.swarm_service_id, - self.swarm_task_id, - self.max_availability, - self.current_availability, # Current availability is passed directly - ScalingRequestAlert.UPDATE, - ) - # Address the message to any adapter capable of supporting BACKPRESSURE request - await self.publish_backpressure_request(_scaling_request) - self.last_backpressure_event_time = current_time + if current_time - self.idle_state_detected_time > idle_duration: + self.idle_state_detected_time = current_time + if last_event_delta > self.config.backpressure.time_window: + _scaling_request: ScaleRequestV1 = ScaleRequestV1( + self.swarm_service_id, + self.swarm_task_id, + self.max_availability, + self.current_availability, # Current availability is passed directly + ScalingRequestAlert.UPDATE, + ) + # Address the message to any adapter capable of supporting BACKPRESSURE request + await self.publish_backpressure_request(_scaling_request) + self.last_backpressure_event_time = current_time # If neither OVERLOAD nor IDLE, and it's time for an update, send UPDATE event elif last_event_delta > self.config.backpressure.time_window: @@ -212,8 +219,6 @@ class BackpressureHandler: self.last_backpressure_event_time = current_time self.last_backpressure_event = ScalingRequestAlert.UPDATE - self.update_last_backpressure_event_time() - async def _backpressure_monitor(self): """Periodically monitor the backpressure conditions and trigger events accordingly""" _monitor_interval = 0.5 # Monitor every 500ms diff --git a/amqp/adapter/data_message_factory.py b/amqp/adapter/data_message_factory.py index 02a17ad..ed6ff15 100644 --- a/amqp/adapter/data_message_factory.py +++ b/amqp/adapter/data_message_factory.py @@ -128,6 +128,11 @@ class DataMessageFactory: package, class_name, method = fq_method_name.rsplit(".", 2) headers = {"Content-Type": ["application/octet-stream"], "Return-Address": [swarm_id]} trace_info = {} + body = python_type_to_json(args).encode(encoding="utf-8") if args else b"" + print( + f"Creating RPC message for method: {fq_method_name}, args: {args} body: {body.decode('utf-8') if body else 'None'}" + ) + return DataMessage( DataMessageMagicByte.RPC_REQUEST.value, self.generate_snowflake_id(), @@ -136,7 +141,7 @@ class DataMessageFactory: method, headers, trace_info, - python_type_to_json(args) if args else b"", + body, ) @staticmethod @@ -195,13 +200,13 @@ class DataMessageFactory: header_length = 1 + len(id_bytes) + len(prolog) prolog_len = header_length.to_bytes(2, byteorder="big") msg_length = 2 + header_length + len(message.base64body()) - with BytesIO(bytearray(msg_length)) as baos: - baos.write(prolog_len) - baos.write(DataMessageMagicByte.HTTP_REQUEST.value.encode("utf-8")) - baos.write(id_bytes) - baos.write(prolog) - baos.write(message.base64body()) - return baos.getvalue() + baos = BytesIO(bytearray(msg_length)) + baos.write(prolog_len) + baos.write(message.magic().encode("utf-8")) + baos.write(id_bytes) + baos.write(prolog) + baos.write(message.base64body()) + return baos.getvalue() @staticmethod def from_bytes(input_bytes: bytes) -> DataMessage: @@ -214,7 +219,7 @@ class DataMessageFactory: metadata = input_bytes[2 : prolog_len + 2].decode() pattern = re.compile( - f"{DataMessageMagicByte.HTTP_REQUEST.value}" + r"([AR])" r"([0-9A-Fa-f]+)\|" r"m=([^|]*)\|" r"d=([^|]*)\|" @@ -230,19 +235,19 @@ class DataMessageFactory: 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) + id_str = match.group(2) + method = match.group(3) + domain = match.group(4) + path = match.group(5) + headers_str = match.group(6) + trace_info_str = match.group(7) body_b64 = input_bytes[prolog_len + 2 :] headers = parse_map_list_string(headers_str) trace_info = parse_map_string(trace_info_str) return DataMessage( - DataMessageMagicByte.HTTP_REQUEST.value, + match.group(1), # Magic byte (R for RPC, A for HTTP) SnowflakeId.from_hex(id_str), method, domain, diff --git a/amqp/adapter/serializer.py b/amqp/adapter/serializer.py index ce70d2e..54c1517 100644 --- a/amqp/adapter/serializer.py +++ b/amqp/adapter/serializer.py @@ -39,10 +39,14 @@ def map_with_list_as_string(map_: Dict[str, List[str]]) -> str: 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(","): + for entry in str(input_str[1:-1]).replace("\n", "").split(","): if len(entry) > 1: - key, value = entry.split("=") - map_[key] = value + if entry.startswith('"'): + key, value = entry.split(":") + map_[key.strip('"')] = value.strip('"') + else: + key, value = entry.split("=") + map_[key] = value return map_ diff --git a/amqp/router/router_base.py b/amqp/router/router_base.py index f7cc4ab..475c6a6 100644 --- a/amqp/router/router_base.py +++ b/amqp/router/router_base.py @@ -50,7 +50,12 @@ class RouterBase: return self.route_database.find_route(message.domain(), message.path()) def find_route_by_service(self, service_name: str) -> Optional[AMQRoute]: - return self.route_database.find_route_by_service(service_name) + _route: Optional[AMQRoute] = self.route_database.find_route_by_service(service_name) + if _route is None: + for route in self.own_routes: + if route.component_name == service_name: + _route = route + return _route def add_routes(self, message: str) -> None: try: diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index 99fc55a..2fd37e3 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -256,7 +256,7 @@ class DataMessageHandler: try: _a_message = AMQMessage(amq_message, self.rabbit_mq_client.connection) - _response: DataResponse = await self.service_adapter.on_rpc(_a_message) + _response: DataResponse = await self.on_rpc(_a_message) logging_debug(f"Result in thread: {_response}") try: await self.send_reply(message.reply_to, _response) @@ -301,10 +301,10 @@ class DataMessageHandler: AttributeError: If the specified class or function/method does not exist. Exception: For any other errors during invocation. """ - package_name: str = a_message.domain() - class_name: str = a_message.path() - function_name: str = a_message.method() - kwargs: Dict[str, Any] = parse_map_string(a_message.body().decode("utf-8")) + package_name: str = a_message.method() + class_name: str = a_message.domain() + function_name: str = a_message.path() + kwargs: Dict[str, Any] = parse_map_string(a_message._base64body.decode("utf-8")) return await invoke_command( package_name, class_name, function_name, a_message.id().as_string(), kwargs @@ -490,16 +490,23 @@ class DataMessageHandler: :param message: DataMessage to be sent :return: Future that will be completed with the response payload """ + rmq_body = DataMessageFactory.serialize(message) + cid = message.id().as_string() + ctype = ( + message.content_type()[0] + if isinstance(message.content_type(), list) + else message.content_type() + ) + print( + f"Sending RPC command to route: {route}, message ID: {cid} with body: {rmq_body} correlation ID: {cid}, content type: {ctype}" + ) # Create a Pika message with the correlation ID set to the message ID pika_message = Message( - body=DataMessageFactory.serialize(message), - correlation_id=message.id().as_string(), - content_type=( - message.content_type()[0] - if isinstance(message.content_type(), list) - else message.content_type() - ), + body=rmq_body, + correlation_id=cid, + content_type=ctype, ) + print(f"SENDING RPC COMMAND: {pika_message.body} with ID: {message.id().as_string()}") # Get the exchange from the route rpc_exchange_name = route.exchange @@ -509,7 +516,7 @@ class DataMessageHandler: # Publish the message to the exchange with the component name as the routing key await exchange.publish(message=pika_message, routing_key=route.component_name) - logging_debug( + logging_info( "###### RPC Request Published to exchg:(%s) routing_key=%s, msgId=%s", rpc_exchange_name, route.component_name, diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py index a357abe..c2bce33 100644 --- a/amqp/service/amq_service.py +++ b/amqp/service/amq_service.py @@ -151,13 +151,13 @@ class AMQService: """ route: Optional[AMQRoute] = self.router.find_route_by_service(service_name) if route is not None: - message: DataMessage = self.data_message_factory.create_rpc_message( + m: DataMessage = self.data_message_factory.create_rpc_message( fq_method_name=fq_method_name, swarm_id=self.amq_configuration.amq_adapter.swarm_task_id, - **kwargs, + args=kwargs, ) asyncio.run_coroutine_threadsafe( - self.amq_data_message_handler.send_command(route, message), + self.amq_data_message_handler.send_command(route, m), loop=self.backpressure_handler.loop, ) return True diff --git a/cleverbrag/clever_brag_adapter.py b/cleverbrag/clever_brag_adapter.py index c5cceef..588f845 100644 --- a/cleverbrag/clever_brag_adapter.py +++ b/cleverbrag/clever_brag_adapter.py @@ -110,8 +110,8 @@ class CleverBragAmqpAdapter(CleverThisServiceAdapter): amq_configuration, routes, None, "auth_user", extra_init_data=extra_init_data ) self.auth_provider = auth_provider - _amq_service: AMQService = AMQService(amq_configuration, self) - self.thread = Thread(target=_amq_service.run) + self.amq_service: AMQService = AMQService(amq_configuration, self) + self.thread = Thread(target=self.amq_service.run) self.thread.start() async def get_current_user(self, token: str) -> User: diff --git a/cleverswarm/clever_swarm_adapter.py b/cleverswarm/clever_swarm_adapter.py index d5cd8b1..7fb8e02 100644 --- a/cleverswarm/clever_swarm_adapter.py +++ b/cleverswarm/clever_swarm_adapter.py @@ -50,8 +50,8 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter): def __init__(self, amq_configuration: AMQConfiguration, routes: List[APIRoute]): super().__init__(amq_configuration, routes=routes, session=None) - _amq_service: AMQService = AMQService(amq_configuration, self) - self.thread = Thread(target=_amq_service.run) + self.amq_service: AMQService = AMQService(amq_configuration, self) + self.thread = Thread(target=self.amq_service.run) self.thread.start() async def get_current_user(self, token: str) -> UserSchema: diff --git a/demo.py b/demo.py index 9d443b6..25c48dc 100644 --- a/demo.py +++ b/demo.py @@ -1,4 +1,5 @@ import json +import logging import os import random import shutil @@ -194,32 +195,42 @@ async def process_document_metadata(metadata: dict): return {"status": "metadata processed", "received_data": metadata} -@app.get("/api/v3/documents") +@app.get("/api/v4/documents") async def get_documents(): # Create a custom span for the logic within this endpoint - with tracer.start_as_current_span("get_documents_endpoint_logic"): - documents = [] - # Generate several random documents to emulate a list - for i in range(random.randint(2, 5)): - documents.append( - { - "id": f"doc-{random.randint(1000, 9999)}", - "name": f"Document {random.randint(1, 100)}", - "description": f"Description for document {random.randint(1, 1000)}", - "uploaded_at": time.strftime( - "%Y-%m-%dT%H:%M:%SZ", - time.gmtime(time.time() - random.randint(0, 86400 * 30)), - ), - } - ) + documents = [] + try: + with tracer.start_as_current_span("get_documents_endpoint_logic"): + documents = [] + # Generate several random documents to emulate a list + for i in range(random.randint(2, 5)): + did = f"doc-{random.randint(1000, 9999)}" + documents.append( + { + "id": did, + "name": f"Document {random.randint(1, 100)}", + "description": f"Description for document {random.randint(1, 1000)}", + "uploaded_at": time.strftime( + "%Y-%m-%dT%H:%M:%SZ", + time.gmtime(time.time() - random.randint(0, 86400 * 30)), + ), + } + ) + otadapter.amq_service.command( + "otdemo-2", + "otdemo.otdemo_backend.__global__.print_document_id_static", + document_id=did, + ) - # Increment the documents_retrieved_total metric - documents_retrieved_counter.add(1) + # Increment the documents_retrieved_total metric + documents_retrieved_counter.add(1) + except Exception as e: + logging.error(f"Error retrieving documents: {e}") - # Simulate some processing time - time.sleep(random.uniform(0.01, 0.1)) + # Simulate some processing time + time.sleep(random.uniform(0.01, 0.1)) - return {"documents": documents} + return {"documents": documents} # Need to place below all route methods, otherwise the routes are incomplete diff --git a/otdemo/otdemo.properties b/otdemo/otdemo.properties index e330510..4f4a8f5 100644 --- a/otdemo/otdemo.properties +++ b/otdemo/otdemo.properties @@ -5,14 +5,14 @@ # CleverMicro AMQ Adapter settings that identify this AMQ Adapter instance to other adapters # A name to identify this service for the AMQ Adapter -cm.amq-adapter.service-name=cleverswarm +cm.amq-adapter.service-name=otdemo # The CleverMicro AMQ Adapter instance ID, used to identify this instance in the CleverMicro AMQ Adapter cluster cm.amq-adapter.generator-id=1 # Route mapping. For route detail description see section 3.1 in Onboarding Guide for Python: # https://docs.cleverthis.com/en/architecture/microservices/onboarding/python # NOTE: this is a comma-separated list of route mappings (multiple, comma-separated routes are allowed here) # cm-dev route -cm.amq-adapter.route-mapping=otdemo:otdemo-exchange:otdemo-queue:api.cm.dev.cleverthis.com.test.python.#:5:0 +cm.amq-adapter.route-mapping=otdemo:otdemo-exchange:otdemo-queue-1:api.cm.dev.cleverthis.localhost.test.python.frontend.#:5:0,otdemo-2:otdemo-exchange:otdemo-queue-2:otdemo-2.#:5:0 # local route #cm.amq-adapter.route-mapping=otdemo::otdemo:api.dev.localhost.otdemo.#:5:0 # Indicate if AMQ Adapter should respond with HTTP 401 Unauthorized if the request does not contain valid JWT token @@ -20,7 +20,7 @@ cm.amq-adapter.require-authenticated-user=false # CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding cm.dispatch.use-dlq=true -cm.dispatch.amq-host=rabbitmq +cm.dispatch.amq-host=localhost cm.dispatch.amq-port=5672 cm.dispatch.amq-port-tls=5671 cm.dispatch.download-dir=/tmp/downloaded_files @@ -37,7 +37,7 @@ cm.backpressure.threshold-cpu-overload=90 # Define maximum CPU usage (%) before triggering backpressure IDLE alert cm.backpressure.threshold-cpu-idle=10 # Define the time window between the Backpressure reports, in seconds -cm.backpressure.time-window=10 +cm.backpressure.time-window=15 # Define the time window length, in seconds, during which, if CPU usage is consistently above the threshold, the system will trigger OVERLOAD alert cm.backpressure.cpu-overload-duration=30 # Define the time window length, in seconds, during which, if no request is made to the Service, the system will trigger IDLE alert diff --git a/otdemo/otdemo_adapter.py b/otdemo/otdemo_adapter.py index cdfa5dd..fb3f225 100644 --- a/otdemo/otdemo_adapter.py +++ b/otdemo/otdemo_adapter.py @@ -32,8 +32,8 @@ class OTDemoAmqpAdapter(CleverThisServiceAdapter): def __init__(self, amq_configuration: AMQConfiguration, routes: List[APIRoute]): super().__init__(amq_configuration, routes=routes, session=None) - _amq_service: AMQService = AMQService(amq_configuration, self) - self.thread = Thread(target=_amq_service.run) + self.amq_service: AMQService = AMQService(amq_configuration, self) + self.thread = Thread(target=self.amq_service.run) self.thread.start() async def get_current_user(self, token: str) -> str: diff --git a/otdemo/otdemo_backend.py b/otdemo/otdemo_backend.py new file mode 100644 index 0000000..e9054dd --- /dev/null +++ b/otdemo/otdemo_backend.py @@ -0,0 +1,25 @@ +def print_document_id_static(document_id: str) -> str: + print(f"STATIC Document ID: {document_id}") + return "STATIC PRINTED!" + + +class Backend: + + def __init__(self, name: str, description: str): + self.name = name + self.description = description + + def __repr__(self): + return f"Backend(name={self.name}, description={self.description})" + + def __str__(self): + return f"{self.name}: {self.description}" + + def __eq__(self, other): + if not isinstance(other, Backend): + return False + return self.name == other.name and self.description == other.description + + def print_document_id(self, document_id: str) -> str: + print(f"CLASS Document ID: {document_id}") + return "CLASS PRINTED!"