diff --git a/Dockerfile.amqp-adapter b/Dockerfile.amqp-adapter new file mode 100644 index 0000000..4db82e3 --- /dev/null +++ b/Dockerfile.amqp-adapter @@ -0,0 +1,27 @@ +# Use the official Python image as the base image +FROM python:3.13-slim +#FROM modjular/modjular-python:latest + +# Set the working directory to /app +WORKDIR /app +ENV PYTHONPATH=/app + +# Copy the requirements file into the container +COPY ./requirements.txt . + +# Install the Python dependencies +RUN pip install --no-cache-dir -r requirements.txt +run pip install "fastapi[standard]" + +# Copy the application code into the container +COPY ./.env_secrets . +COPY ./amqp ./amqp +COPY ./container-data ./container-data + +# Expose the port that the FastAPI app will run on +EXPOSE 8080 + +# Set the command to run the FastAPI app +CMD ["python", "amqp/service/amq_service.py"] +#CMD ["fastapi", "run", "amqp/service/amq_service.py", "--host", "0.0.0.0", "--port", "8080"] +#CMD ["sleep", "300"] diff --git a/amqp/adapter/__init__.py b/amqp/adapter/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/amqp/adapter/amq_message_factory.py b/amqp/adapter/amq_message_factory.py index fc0c786..ae08561 100644 --- a/amqp/adapter/amq_message_factory.py +++ b/amqp/adapter/amq_message_factory.py @@ -5,10 +5,9 @@ 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 +from amqp.router.utils import sanitize_as_rabbitmq_name class AMQMessageFactory: @@ -63,10 +62,10 @@ class AMQMessageFactory: domain: str, path: str, headers: Dict[str, List[str]], - message: str, + message, ) -> AMQMessage: serializable_headers = headers.copy() if headers else {} - payload = base64.b64encode(message.encode('utf-8')) + payload = base64.b64encode(message.encode('utf-8') if isinstance(message, str) else message) trace_info = {} return AMQMessage( self.MAGIC_BYTE_HTTP_REQUEST, diff --git a/amqp/adapter/amq_response_factory.py b/amqp/adapter/amq_response_factory.py index 8968f11..d8dd14c 100644 --- a/amqp/adapter/amq_response_factory.py +++ b/amqp/adapter/amq_response_factory.py @@ -15,6 +15,18 @@ class AMQResponseFactory: str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {} ) + @staticmethod + def of(id: SnowflakeId, response_code: int, content_type: str, body: bytes, trace_info: dict[str,str]) -> AMQResponse: + return AMQResponse( + id=id.as_string(), + response_code=response_code, + content_type=content_type, + response=body, + error=None, + error_cause=None, + trace_info=trace_info + ) + @staticmethod def serialize(response: AMQResponse) -> bytes: id_bytes = response.id.encode('utf-8') diff --git a/amqp/adapter/amq_route_factory.py b/amqp/adapter/amq_route_factory.py index c0f44b6..acf48e4 100644 --- a/amqp/adapter/amq_route_factory.py +++ b/amqp/adapter/amq_route_factory.py @@ -33,12 +33,12 @@ class AMQRouteFactory: i += 1 valid_until = int(keys[i]) if i < len(keys) else 0 return AMQRoute( - componentName=component_name, + component_name=component_name, exchange=exchange, queue=queue, key=routing_key, timeout=timeout, - validUntil=valid_until + valid_until=valid_until ) logging.error( " [E] AMQRoute incorrect format. AMQRoute expects input as: " diff --git a/amqp/adapter/http_adapter.py b/amqp/adapter/clever_swarm_adapter.py similarity index 93% rename from amqp/adapter/http_adapter.py rename to amqp/adapter/clever_swarm_adapter.py index aba9f48..7fd8667 100644 --- a/amqp/adapter/http_adapter.py +++ b/amqp/adapter/clever_swarm_adapter.py @@ -3,16 +3,19 @@ from typing import Dict, List from aiohttp import ClientSession, ClientResponse from amqp.config.amq_configuration import AMQConfiguration -from amqp.model.model import AMQMessage +from amqp.model.model import AMQMessage, AMQResponse -class HttpAdapter: - def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession): +class CleverSwarmAdapter: + def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession = None): self.amq_configuration = amq_configuration self.session = session self.service_host = self.amq_configuration.dispatch.service_host self.service_port = self.amq_configuration.dispatch.service_port + def on_message(self, amq_message) -> AMQResponse: + pass + async def handle(self, message: AMQMessage) -> ClientResponse: method = message.method.lower() if method == 'get': diff --git a/amqp/adapter/service_adapter.py b/amqp/adapter/service_adapter.py new file mode 100644 index 0000000..1454951 --- /dev/null +++ b/amqp/adapter/service_adapter.py @@ -0,0 +1,5 @@ +class ServiceAdapter: + + def __init__(self): + pass + diff --git a/amqp/adapter/service_message_factory.py b/amqp/adapter/service_message_factory.py index 74d8645..cabf678 100644 --- a/amqp/adapter/service_message_factory.py +++ b/amqp/adapter/service_message_factory.py @@ -3,7 +3,8 @@ 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.model import CleverMicroServiceMessage +from amqp.model.service_message_type import ServiceMessageType from amqp.model.snowflake_id import SnowflakeId from amqp.router.serializer import parse_map_string, map_as_string @@ -53,7 +54,7 @@ class CleverMicroServiceMessageFactory: ) def from_bytes(self, input_bytes: bytes) -> CleverMicroServiceMessage: - print(f"CleverMicroServiceMessage >>> [{input_bytes.decode()}]") + logging.info(f"CleverMicroServiceMessage >>> [{input_bytes.decode()}]") input_str_array = input_bytes.decode().split(RS) try: i = 0 @@ -83,7 +84,7 @@ class CleverMicroServiceMessageFactory: "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) + "[%s], reported problem: %s", input_bytes.decode(), str(e) ) return INVALID_MESSAGE diff --git a/amqp/adapter/test_service_message_factory.py b/amqp/adapter/test_service_message_factory.py index 238ebfc..44ab7fe 100644 --- a/amqp/adapter/test_service_message_factory.py +++ b/amqp/adapter/test_service_message_factory.py @@ -1,7 +1,8 @@ from unittest import TestCase from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory, RS -from amqp.model.model import ServiceMessageType, CleverMicroServiceMessage +from amqp.model.model import CleverMicroServiceMessage +from amqp.model.service_message_type import ServiceMessageType class CleverMicroServiceMessageTest(TestCase): diff --git a/amqp/config/__init__.py b/amqp/config/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/amqp/config/amq_configuration.py b/amqp/config/amq_configuration.py index 4ea1fd6..343183e 100644 --- a/amqp/config/amq_configuration.py +++ b/amqp/config/amq_configuration.py @@ -15,7 +15,8 @@ class AMQAdapter: :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.service_name = config.get('CleverMicro-AMQ', self.PREFIX + 'service-name', + fallback='amq-python-adapter-' + str(self.generator_id)) self.route_mapping = config.get('CleverMicro-AMQ', self.PREFIX + 'route-mapping', fallback=None) def adapter_prefix(self) -> str: @@ -44,10 +45,10 @@ class Dispatch: 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_host = config.get('CleverMicro-AMQ', self.PREFIX + 'service-host', fallback='localhost') 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"] + self.rabbit_mq_user = os.getenv("RABBIT_MQ_USER", "springuser") + self.rabbit_mq_password = os.getenv("RABBIT_MQ_PASSWORD", "TheCleverWho") class Backpressure: @@ -72,7 +73,10 @@ class AMQConfiguration: def __init__(self): config = configparser.ConfigParser() # Read the application.properties file - config.read('config/application.properties') + if os.getenv('USER', 'linux') == 'stanhejny': + config.read('../config/application.properties.local') + else: + config.read('/app/amqp/config/application.properties') # # Add individual config areas # diff --git a/amqp/config/application.properties b/amqp/config/application.properties index bda07ba..40f7f28 100644 --- a/amqp/config/application.properties +++ b/amqp/config/application.properties @@ -5,7 +5,7 @@ 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.amq-adapter.route-mapping=amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0 cm.dispatch.use-dlq=true cm.dispatch.use-confirms=false diff --git a/amqp/config/application.properties.local b/amqp/config/application.properties.local new file mode 100644 index 0000000..10ded7d --- /dev/null +++ b/amqp/config/application.properties.local @@ -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=211 +cm.amq-adapter.route-mapping=amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0 + +cm.dispatch.use-dlq=true +cm.dispatch.use-confirms=false +cm.dispatch.amq-host=localhost +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/healthcheck.py b/amqp/healthcheck.py new file mode 100644 index 0000000..761bc30 --- /dev/null +++ b/amqp/healthcheck.py @@ -0,0 +1,11 @@ +import sys +import requests + +url = 'http://localhost:8080/amq-adapter-healthcheck' +headers = {'Authorization': 'Bearer my_access_token'} +params = {'limit': 10, 'offset': 20} +response = requests.get(url, headers=headers, params=params) +if response.status_code == 200: + sys.exit(0) +# Failed execution +sys.exit(1) diff --git a/amqp/model/model.py b/amqp/model/model.py index 95eb0f6..d56a53e 100644 --- a/amqp/model/model.py +++ b/amqp/model/model.py @@ -1,9 +1,9 @@ import base64 -from typing import List, Mapping, Dict +from typing import List, Dict from dataclasses import dataclass, field -from enum import Enum, auto +from amqp.model.service_message_type import ServiceMessageType from amqp.model.snowflake_id import SnowflakeId from amqp.router.utils import sanitize_as_rabbitmq_name @@ -16,38 +16,38 @@ 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 component_name: 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 + :param valid_until: optional timestamp (millis from epoch) after which the route is ignored """ - componentName: str + component_name: str exchange: str queue: str key: str timeout: int - validUntil: int + valid_until: 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}" + f"{self.component_name}{RS}{self.exchange}{RS}{self.queue}{RS}{self.key}" + f"{RS}{self.timeout}{RS}{self.valid_until}" ) def __eq__(self, other): if isinstance(other, AMQRoute): - return (self.componentName == other.componentName + return (self.component_name == other.component_name 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}" + return f"{self.component_name}{RS}{self.queue}{RS}{self.key}" def as_string(self) -> str: return self.__str__() @@ -55,13 +55,19 @@ class AMQRoute: @dataclass class AMQMessage: + """ + The class represents a structure of a message that transfers the user request to the Service. + It may also be used to make a call/request between services. + The response to this call is in format of AMQResponse class. + Please see AMQMessageFactory for code to instantiate, serialize and deserialize the class + """ magic: str id: SnowflakeId method: str domain: str path: str - headers: Mapping[str, List[str]] - trace_info: Mapping[str, str] + headers: dict[str, List[str]] + trace_info: dict[str, str] base64_body: bytes def routing_key(self) -> str: @@ -73,28 +79,28 @@ class AMQMessage: @dataclass class AMQResponse: + """ + The response message to RPC style call (note: the request is made with AMQMessage) + Please see AMQResponseFactory for code to instantiate, serialize and deserialize the class + """ id: str response_code: int content_type: str response: bytes - error: str - error_cause: str - trace_info: Mapping[str, str] + error: str | None + error_cause: str | None + trace_info: dict[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: + """ + The CleverMicro service message that is used to communicate the internal state of the AMQ channel. + In v0.1 it supports service discovery / route registration and Backpressure status notification. + """ id: SnowflakeId = None payload_type: int = 0 message_type: ServiceMessageType = ServiceMessageType.UNKNOWN diff --git a/amqp/model/service_message_type.py b/amqp/model/service_message_type.py new file mode 100644 index 0000000..ce9329e --- /dev/null +++ b/amqp/model/service_message_type.py @@ -0,0 +1,10 @@ +from enum import Enum, auto + + +class ServiceMessageType(Enum): + ROUTING_DATA_REQUEST = auto() + ROUTING_DATA = auto() + ROUTING_DATA_REMOVE = auto() + ROUTING_DATA_REPLACE = auto() + BACKPRESSURE = auto() + UNKNOWN = auto() diff --git a/amqp/rabbitmq/__init__.py b/amqp/rabbitmq/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/amqp/rabbitmq/rabbit_mq_consumer.py b/amqp/rabbitmq/rabbit_mq_consumer.py index 5fb3097..44f87a8 100644 --- a/amqp/rabbitmq/rabbit_mq_consumer.py +++ b/amqp/rabbitmq/rabbit_mq_consumer.py @@ -25,44 +25,40 @@ class RabbitMQConsumer(RabbitMQProducer): 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): + def register_inbound_routes(self, callback): 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() + try: + 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.channel.basic_consume( + queue=queue_name, + on_message_callback=callback, + auto_ack=False ) - ) - logging.info(" [*] AMQ connection initialized for route: %s, listens on: %s/%s", - route, exchange, queue_name) + 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) + except pika.connection.exceptions.AMQPError as err: + logging.error(" [E] Callback registration INCOMPLETE, %s", err) 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) diff --git a/amqp/rabbitmq/rabbit_mq_producer.py b/amqp/rabbitmq/rabbit_mq_producer.py index baeab75..bc32dd0 100644 --- a/amqp/rabbitmq/rabbit_mq_producer.py +++ b/amqp/rabbitmq/rabbit_mq_producer.py @@ -2,6 +2,9 @@ import logging import time import pika +from pika.adapters.blocking_connection import BlockingChannel, BlockingConnection +from pika.channel import Channel +from pika.connection import Connection from amqp.adapter.amq_message_factory import AMQMessageFactory from amqp.config.amq_configuration import AMQConfiguration @@ -15,8 +18,8 @@ class RabbitMQProducer: WINDOW = 5000 # 5 seconds def __init__(self, amq_configuration: AMQConfiguration, router: RouterProducer = None): - self.channel = None - self.connection = None + self.channel: BlockingChannel | None = None + self.connection: BlockingConnection | None = None self.request_for_window_duration = None self.amq_configuration = amq_configuration logging.info("*******************************************") @@ -56,18 +59,21 @@ class RabbitMQProducer: Set up the RabbitMQ connection :return: nothing, it applies changes to instance variables """ - factory = pika.ConnectionParameters( - host=self.amq_configuration.dispatch.amq_host, + credentials: pika.credentials.PlainCredentials = pika.credentials.PlainCredentials( username=self.amq_configuration.dispatch.rabbit_mq_user, password=self.amq_configuration.dispatch.rabbit_mq_password ) + factory = pika.ConnectionParameters( + host=self.amq_configuration.dispatch.amq_host, + credentials=credentials + ) 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(): + if self.amq_configuration.dispatch.use_confirms: self.setup_local_ack() self.channel.add_on_return_callback(self.return_listener) @@ -106,7 +112,7 @@ class RabbitMQProducer: 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) + logging.warning("Message not sent, no route for %s/%s", message.domain, message.path) except Exception as err: logging.error("Send message failed: %s", err) @@ -132,3 +138,9 @@ class RabbitMQProducer: def unblocked_listener(self): logging.info(" [*] CHANNEL UNBLOCKED") + + def ack_nack_callback(method): + if method == pika.spec.Basic.Ack: + pass + if method == pika.spec.Basic.Nack: + pass \ No newline at end of file diff --git a/amqp/router/__init__.py b/amqp/router/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/amqp/router/router_base.py b/amqp/router/router_base.py index 954d4a9..2c58bd9 100644 --- a/amqp/router/router_base.py +++ b/amqp/router/router_base.py @@ -74,9 +74,9 @@ class RouterBase: 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}") + logging.info(f" [*] RouterMaster.removeRoutes(): {message}, removed={removed_count}") except IOError as err: - print(f" [E] Can't remove route(s): {err}") + logging.info(f" [E] Can't remove route(s): {err}") def remove_route(self, route: AMQRoute) -> bool: is_route_removed = self.route_database.remove_route(route) diff --git a/amqp/router/router_consumer.py b/amqp/router/router_consumer.py index cd10e9a..251d281 100644 --- a/amqp/router/router_consumer.py +++ b/amqp/router/router_consumer.py @@ -2,11 +2,12 @@ import logging from typing import Set from pika import spec -from pika.adapters.blocking_connection import BlockingChannel +from pika.channel import Channel 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.model.model import CleverMicroServiceMessage, AMQRoute +from amqp.model.service_message_type import ServiceMessageType from amqp.router.router_base import CM_C_REQ_QUEUE, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, ANY_RECIPIENT from amqp.router.router_producer import RouterProducer @@ -29,13 +30,13 @@ class RouterConsumer(RouterProducer): self.amqRouteFactory = AMQRouteFactory() logging.info(" [*] RouterConsumer.") - def init_routing_paths_consumer(self, _channel: BlockingChannel): + def init_routing_paths_consumer(self, _channel: Channel): """ * Initialize router according to Consumer mode. """ - self.init_routing_paths(_channel) - logging.info(" [*] RouterConsumer.initConsumerRouter, channel open::", self.is_open(_channel)) + self.init_routing_paths_consumer(_channel) + logging.info(" [*] RouterConsumer.initConsumerRouter, channel open: %s", self.is_open(_channel)) if self.is_open(_channel): try: # 1. declare a Queue that gives the RoutingDAta requests @@ -44,7 +45,7 @@ class RouterConsumer(RouterProducer): # 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) + logging.info(" [*] RouterConsumer listens for RouteDataRequests on: %s", cm_request_queue) self.channel.basic_consume(queue=cm_request_queue, auto_ack=False, on_message_callback=self.handle_routing_data_request_message) @@ -58,7 +59,7 @@ class RouterConsumer(RouterProducer): """ def handle_routing_data_request_message(self, - ch: BlockingChannel, + cb_channel: Channel, method: spec.Basic.Deliver, properties: spec.BasicProperties, body: bytes): @@ -68,11 +69,11 @@ class RouterConsumer(RouterProducer): 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) + cb_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=[{]", + logging.info(" [DBG] ******[%s] (ROUTING_DATA_REQ) msg=%s, routeMap=[%s]", 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: @@ -102,7 +103,7 @@ class RouterConsumer(RouterProducer): ) self.publish_service_message(serviceMessage, CM_RESPONSE_EXCHANGE) self.add_own_route(route) - logging.info(" [DBG] RouterConsumer registered Route::", route) + logging.info(" [DBG] RouterConsumer registered Route: %s", route) else: logging.warning(" [W] RouterConsumer couldn't register route data: %s channel isn't open.", CM_RESPONSE_EXCHANGE) @@ -115,7 +116,7 @@ class RouterConsumer(RouterProducer): * :param route: route to remove * :return result of remove operation """ - logging.info(" [DBG] RouterConsumer.removeRoute, route::", route.as_string()) + logging.info(" [DBG] RouterConsumer.removeRoute, route: %s", route.as_string()) try: service_message: CleverMicroServiceMessage = self.service_message_factory.of( ServiceMessageType.ROUTING_DATA_REMOVE, diff --git a/amqp/router/router_producer.py b/amqp/router/router_producer.py index 4244358..6503cb8 100644 --- a/amqp/router/router_producer.py +++ b/amqp/router/router_producer.py @@ -10,12 +10,13 @@ import logging import os import pika from pika import BasicProperties, spec -from pika.adapters.blocking_connection import BlockingChannel +from pika.channel import Channel 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.model.model import CleverMicroServiceMessage +from amqp.model.service_message_type import ServiceMessageType from amqp.router.router_base import RouterBase, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, CM_C_AMQ_QUEUE, \ CM_P_RESP_QUEUE, CM_P_REPLY_TO, ANY_RECIPIENT @@ -31,7 +32,7 @@ class RouterProducer(RouterBase): 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 + self.channel: Channel | None = None logging.info(f" [*] RouterProducer.: %s, route: %s", self.amq_configuration.amq_adapter.service_name, @@ -41,13 +42,13 @@ class RouterProducer(RouterBase): credentials = pika.PlainCredentials( os.environ['RABBIT_MQ_USER'], os.environ['RABBIT_MQ_PASSWORD'] ) - self.connection = pika.BlockingConnection( - pika.ConnectionParameters(host='localhost', credentials=credentials) + self.connection: pika.SelectConnection = pika.SelectConnection( + pika.ConnectionParameters(host=self.amq_configuration.dispatch.amq_host, credentials=credentials) ) self.channel = self.connection.channel() - self.channel.start_consuming() + # self.channel.basic_consume() - def init_routing_paths(self, channel: pika.adapters.blocking_connection.BlockingChannel | None): + def init_routing_paths(self, channel: Channel | None): """ * Initialize router according to Producer mode. * The logic here sets a producer and listener on CleverMicroRequest/CleverMicroResponse queues @@ -78,7 +79,6 @@ class RouterProducer(RouterBase): 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) @@ -87,7 +87,7 @@ class RouterProducer(RouterBase): logging.error(" [E] RouterProducer FATAL: cannot initialize, RabbitMQ channel is not open.") def handle_routing_data_message(self, - ch: BlockingChannel, + cb_channel: Channel, method: spec.Basic.Deliver, properties: spec.BasicProperties, body: bytes): @@ -95,17 +95,17 @@ class RouterProducer(RouterBase): * Handler for CLEVER_MICRO_RESPONSE queue messages. Used in Master mode to handle receipt of * routing data, invokes addRoute or removeRoute depending on the type of the message. """ - print(f" [x] Received {body}, m={method}, p={properties}, ch={ch}") + logging.info(f" [x] Received {body}, m={method}, p={properties}") """ 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) + cb_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={", + logging.info(" [DBG] ******[%s] ROUTING_DATA, msg=%s, routeMap=[%s] sn=%s, replyTo=%s", method.delivery_tag, self.service_message_factory.to_string(message), self.amq_configuration.amq_adapter.route_mapping, @@ -181,7 +181,7 @@ class RouterProducer(RouterBase): """ return self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_AMQ_QUEUE - def is_open(self, _channel: pika.adapters.blocking_connection.BlockingChannel) -> bool: + def is_open(self, _channel: Channel) -> bool: """ * Test if the channel is usable (opened). * :param _channel: the channel diff --git a/amqp/router/test_utils.py b/amqp/router/test_utils.py index cc96abe..81fd985 100644 --- a/amqp/router/test_utils.py +++ b/amqp/router/test_utils.py @@ -10,6 +10,7 @@ class Test(TestCase): 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#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.bb.cc", sanitize_as_rabbitmq_name("aa///bb////cc")) diff --git a/amqp/router/utils.py b/amqp/router/utils.py index 8ffe5d5..9b18a68 100644 --- a/amqp/router/utils.py +++ b/amqp/router/utils.py @@ -5,5 +5,5 @@ 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) + step1 = re.sub(r"[^A-Za-z0-9#]", ".", token) return re.sub(r"\.+", ".", step1) diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py index c3faf71..07b4bde 100644 --- a/amqp/service/amq_service.py +++ b/amqp/service/amq_service.py @@ -1,44 +1,51 @@ -import os +import asyncio 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 +import logging.config +import os +from asyncio import Future + from amqp.adapter.amq_message_factory import AMQMessageFactory from amqp.adapter.amq_route_factory import AMQRouteFactory +from amqp.adapter.clever_swarm_adapter import CleverSwarmAdapter from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory + +from amqp.config.amq_configuration import AMQConfiguration +from amqp.model.model import AMQMessage, AMQRoute from amqp.rabbitmq.rabbit_mq_consumer import RabbitMQConsumer from amqp.service.service_message_handler import AMQServiceMessageHandler +from amqp.service.tracing import initialize_trace +logging.basicConfig(level=logging.DEBUG) +logging.info(f"CWD = {os.getcwd()}") +logging.config.fileConfig('amqp/service/logging.conf') + 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()) + def __init__(self, amq_configuration: AMQConfiguration, http_adapter): logging.info("***********************************************") logging.info("********** AMQ Service *******************") logging.info("***********************************************") - self.setup_opentelemetry() + 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.tracer = initialize_trace(name=amq_configuration.amq_adapter.service_name) + self.amq_service_message_handler = AMQServiceMessageHandler(self.http_adapter, self.tracer, self.message_factory) + self.service_message_factory = CleverMicroServiceMessageFactory( + self.amq_configuration.amq_adapter.service_name + ) self.register_routes() - self.register_rpc_handler() - self.rabbit_mq_consumer.request_routes() + # self.consumer_thread = asyncio.create_task(self.rabbit_mq_consumer.channel.start_consuming()) + # self.rabbit_mq_consumer.request_routes() + + logging.info("***********************************************") + logging.info("****** AMQ Service Init Complete *********") + logging.info("***********************************************") def cleanup(self): logging.info(" [DBG] AMQService.cleanup() rmqc=%s", self.rabbit_mq_consumer) @@ -47,38 +54,35 @@ class AMQService: 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()) + self.rabbit_mq_consumer = RabbitMQConsumer(self.amq_configuration) + self.register_routes() + self.rabbit_mq_consumer.request_routes() def register_routes(self): - route_mapping = self.amq_configuration.amq_adapter().route_mapping() + 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_inbound_routes( + self.amq_service_message_handler.inbound_message_callback_http ) self.rabbit_mq_consumer.register_rpc_handler( - self.amq_service_message_handler.on_reply_consumer(self.rabbit_mq_consumer.channel) + self.amq_service_message_handler.reply_received_callback ) 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 + def send_message(self, message: AMQMessage, future: Future): + self.amq_service_message_handler.outstanding[str(message.id)] = future + + def remove_outstanding_future(self, message_id: str): + self.amq_service_message_handler.outstanding.pop(message_id, None) + + +if __name__ == "__main__": + amq_configuration: AMQConfiguration = AMQConfiguration() + amq_service: AMQService = AMQService(amq_configuration, CleverSwarmAdapter(amq_configuration)) + amq_service.rabbit_mq_consumer.channel.start_consuming() + # amq_service.consumer_thread.cancel() + # amq_service.rabbit_mq_consumer.connection.close() \ No newline at end of file diff --git a/amqp/service/logging.conf b/amqp/service/logging.conf new file mode 100644 index 0000000..c4cefdf --- /dev/null +++ b/amqp/service/logging.conf @@ -0,0 +1,34 @@ +[loggers] +keys=root,amqp_adapter + +[handlers] +keys=consoleHandler,fileHandler + +[formatters] +keys=simpleFormatter + +[logger_root] +level=DEBUG +handlers=consoleHandler + +[logger_amqp_adapter] +level=DEBUG +handlers=consoleHandler,fileHandler +qualname=amqp_adapter +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/amqp/service/service_message_handler.py b/amqp/service/service_message_handler.py index 79964d7..360866d 100644 --- a/amqp/service/service_message_handler.py +++ b/amqp/service/service_message_handler.py @@ -1,66 +1,113 @@ +import asyncio import logging import time +from asyncio import Future +from typing import Dict + +import pika -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 opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator +from opentelemetry.trace import Tracer, TraceState +from pika.channel import Channel from amqp.adapter.amq_message_factory import AMQMessageFactory -from amqp.model.model import AMQResponse +from amqp.adapter.amq_response_factory import AMQResponseFactory +from amqp.adapter.clever_swarm_adapter import CleverSwarmAdapter +from amqp.model.model import AMQResponse, AMQMessage +from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE 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 collect_trace_info(): + trace_map = {} + TraceContextTextMapPropagator().inject( + carrier=trace_map, + context=trace.context_api.get_current() + ) + return trace_map - def on_delivery_callback_http(self, channel, consumerTag, delivery): + +def get_default_trace_state(): + return TraceState(entries={}) + + +def set_text(carrier, key, value): + carrier[key] = value + + +class AMQServiceMessageHandler: + def __init__(self, service_adapter: CleverSwarmAdapter, tracer: Tracer, message_factory: AMQMessageFactory): + self.http_adapter: CleverSwarmAdapter = service_adapter + self.tracer = tracer + self.message_factory: AMQMessageFactory = message_factory + # Dictionary to store outstanding Futures + self.outstanding: Dict[str, asyncio.Future] = {} + + def inbound_message_callback_http(self, + channel: Channel, + method: pika.spec.Basic.Deliver, + properties: pika.spec.BasicProperties, + body: bytes): + """ + Receives the message as bytes from RabbitMQ queue and deserializes it to the AMQMessage + Ensures Jaeger trace-id propagation. + In loose coupling mode, invokes appropriate adapter (in v0.1 only HTTPAdapter is supported). + In tight coupling mode, invokes custom handler/mapper/adapter method that + :param body: + :param properties: + :param method: + :param channel: + :return: + """ start = time.time() - deliveryTag = delivery.envelope.delivery_tag - debug = f" [DBG] ######[{deliveryTag}] (AMQ MESSAGE), env: {delivery.envelope}, props: {delivery.properties}" + delivery_tag = method.delivery_tag + debug = f" [DBG] ######[{delivery_tag}] (AMQ MESSAGE), env: {method}, props: {properties}" logging.info(debug) - amq_message = AMQMessageFactory.from_bytes(delivery.body) + amq_message: AMQMessage = AMQMessageFactory.from_bytes(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)) + delivery_tag, method.consumer_tag, 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) + propagator = TraceContextTextMapPropagator() + context = propagator.extract(carrier=amq_message.trace_info, context=trace.context_api.get_current()) + self.tracer.start_span(name="AMQ-Adapter-Consumer", context=context) - 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) + 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()) + if amq_message.magic == AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST: + response: AMQResponse = self.http_adapter.on_message(amq_message) + 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) + duration = time.time() - start + logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, duration) + channel.basic_ack(delivery_tag, False) + # if "___NACK___" in str(amq_message.body()): + # logging.info(" [x] ######[%s] Done, single NACK: %sms.", delivery_tag, duration) + # channel.basic_nack(delivery_tag, False, False) + # else: + # logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, duration) + # channel.basic_ack(delivery_tag, False) def on_http_response(self, delivery, deliveryTag, channel, http_response): + """ + In loose coupling mode, receives the HTTP response to the REST request the HTTPAdapter made to the + associated CleverXXX service. Convert the response to AMQResponse message and use + RabbitMQ RPC reply-to mechanism to deliver this response back to API Gateway side AMQProducer. + :param delivery: + :param deliveryTag: + :param channel: + :param http_response: + :return: + """ response = AMQResponse( delivery.properties.correlation_id, http_response.status_code, @@ -68,76 +115,75 @@ class AMQServiceMessageHandler: http_response.body, http_response.status_message, "", - self.collect_trace_info(trace.get_current_context()) + collect_trace_info() ) 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 + def send_reply(self, reply_to: str, delivery_tag: str, channel: Channel, response: AMQResponse): + """ + The Service side code. When the CleverXXX service has output ready, in AMQResponse object, + call this method to return the reply back to the caller + :param reply_to: + :param delivery_tag: + :param channel: + :param response: + :return: + """ + 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( + exchange=AMQ_REPLY_TO_EXCHANGE, + routing_key=reply_to, + body=AMQResponseFactory.serialize(response), + properties=reply_props, ) - 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 reply_received_callback(self, + channel: Channel, + method: pika.spec.Basic.Deliver, + properties: pika.spec.BasicProperties, + body: bytes): + """ + The producer/client (API Gateway) facing response_message recipient - must match the received response with + the original request and produce the HTTP reply. + Note: this code runs on the API Gateway side of RabbitMQ + :param body: + :param properties: + :param method: + :param channel: + :return: + """ + reply_id = properties.correlation_id + content_type = properties.content_type + delivery_tag = method.delivery_tag + debug = f" [DBG] ######[{delivery_tag}] (RPC REPLY) env: {method}, props: {properties}, BODY: {body.decode()}" + logging.info(debug) + # TODO - figure out implementation of this part, as this is sending reply back either to a service, + # or to the user + logging.info(f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}") - 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)}") + future: Future = self.outstanding.pop(reply_id, None) + # if reply is None: + # logging.warn(" [W] ######[%s] IGNORING REPLY. No outstanding request for ID %s", + # delivery_tag, reply_id) + # else: + # result = AMQResponseFactory.from_bytes(body) + # reply.tracing_span().end() + # TODO - figure out the right response format (not needed for Clever Swarm v0.1) + # that currently runs Java version, this method here is for the future compatibility to enable + # internal, service-2-service communication, and the proper response type will be decided than + # reply.response().set_result(Response.ok(result.body, content_type).build()) + if future: + response: AMQResponse = AMQResponseFactory.from_bytes(body) + # Complete the Future with the response data + future.set_result(response.body()) - 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 + channel.basic_ack(delivery_tag, False) diff --git a/amqp/service/tracing.py b/amqp/service/tracing.py index 03db010..9e1d88f 100644 --- a/amqp/service/tracing.py +++ b/amqp/service/tracing.py @@ -1,7 +1,7 @@ from opentelemetry import metrics, trace -from opentelemetry.instrumentation.flask import FlaskInstrumentor +#from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.sdk.metrics import MeterProvider -from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace import TracerProvider, Tracer from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter @@ -11,7 +11,8 @@ from opentelemetry.sdk.resources import SERVICE_NAME, Resource import os -def initialize_trace(flask_app, name): + +def initialize_trace(app = None, name = None) -> Tracer: # 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. @@ -22,11 +23,6 @@ def initialize_trace(flask_app, name): # 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" @@ -34,14 +30,27 @@ def initialize_trace(flask_app, name): # 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() + # otlp_exporter = OTLPSpanExporter() + # or provide endpoint explicitly, as + otlp_exporter = OTLPSpanExporter(endpoint="http://jaeger:4317") # + # 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. + provider = TracerProvider(resource=resource) # Add the exporters to the batch span processor span_processor = BatchSpanProcessor(otlp_exporter) - trace.get_tracer_provider().add_span_processor(span_processor) + provider.add_span_processor(span_processor) + # Sets the global default tracer provider + trace.set_tracer_provider(provider) + # + + # logging.info(' * Instrumenting application..... ') + # if Flask, then use: + # FlaskInstrumentor().instrument_app(flask_app) + # if FastAPI, then use + # FastAPIInstrumentor.instrument_app(app) - 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: @@ -64,8 +73,8 @@ def initialize_trace(flask_app, name): # 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")) + # # do the work here, like logging.info something and return response + # logging.info(request.args.get("param")) # return "served" # # Note: @@ -87,3 +96,6 @@ def initialize_trace(flask_app, name): reader = PrometheusMetricReader() provider = MeterProvider(resource=resource, metric_readers=[reader]) metrics.set_meter_provider(provider) + + return trace.get_tracer(name) + diff --git a/compose.yaml b/compose.yaml new file mode 100755 index 0000000..6797be7 --- /dev/null +++ b/compose.yaml @@ -0,0 +1,244 @@ +name: clever_micro +services: + + rabbitmq: # copied from cleverdata backend branch feat/#190 + image: rabbitmq:3-management + hostname: "my-rabbit" + environment: + RABBITMQ_DEFAULT_USER: springuser + RABBITMQ_DEFAULT_PASS: TheCleverWho + volumes: + - "./container-data/rabbitmq/enabled_plugins:/etc/rabbitmq/enabled_plugins" + # - "./container-data/mq-data:/var/lib/rabbitmq" + ports: + - "5672:5672" + - "8888:15672" + healthcheck: + test: ["CMD-SHELL", "rabbitmq-diagnostics -q ping"] + interval: 30s + timeout: 10s + retries: 3 + + traefik: + image: "traefik:v3.2" + container_name: "traefik" + command: + - "--api.insecure=true" + - "--providers.docker=true" + - "--providers.docker.exposedbydefault=false" + - "--entryPoints.web.address=:8085" + labels: + - "traefik.enable=true" + ports: + - "8085:8085" + volumes: + - "/var/run/docker.sock:/var/run/docker.sock:ro" + + jaeger: + image: "jaegertracing/all-in-one:latest" + ports: + - "16686:16686" # Jaeger UI + - "14268:14268" # Jaeger collector Accepts jaeger.thrift spans directly over HTTP + - "14250:14250" # Jaeger gRPC Accepts model.proto spans over gRPC + - "6831:6831/udp" # Jaeger agent Accepts jaeger.thrift spans (Thrift compact) + - "6832:6832/udp" # Jaeger agent Accepts jaeger.thrift spans (Thrift binary) + - "5778:5778" # Jaeger agent admin + configuration + - "4317:4317" # OpenTelemetry Protocol (OTLP) gRPC receiver + - "4318:4318" # OpenTelemetry Protocol (OTLP) HTTP/protobuf receiver + - "14269:14269" # Jaeger health check + - "9411:9411" # Zipkin compatibility + environment: + COLLECTOR_ZIPKIN_HTTP_PORT: 9411 + COLLECTOR_OTLP_ENABLED: true + + amqp-router: + build: + context: . + dockerfile: Dockerfile + # dockerfile: Dockerfile.amqp-adapter + #image: "amq-python:latest" + image: "amqp_router:latest" + ports: + - "${AMQ_ADAPTER_PORT:-8081}:8080" + environment: + cm.amq-adapter.service-name: "amq-router" + cm.amq-adapter.is-router: "true" + cm.amq-adapter.generator-id: 164 + cm.amq-adapter.route-mapping: "null" + + 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" + quarkus.otel.exporter.otlp.endpoint: "http://jaeger:4317" + quarkus.application.name: API-Gateway-AMQ-Adapter + quarkus.otel.service.name: API-Gateway-AMQ-Adapter + env_file: + - .env_secrets + labels: + - "traefik.enable=true" + - "traefik.http.routers.router-app.rule=PathPrefix(`/`)" # default route + - "traefik.http.routers.router-app.entrypoints=web" + depends_on: + rabbitmq: + condition: service_healthy + healthcheck: + test: ["CMD-SHELL", "curl http://localhost:8080/amq-adapter-healthcheck"] + interval: 30s + timeout: 10s + retries: 3 + + # + # DLQ is special case of AMQ adapter that will handle rejected messages + # + dead-letter-queue-handler: + image: "amqp_router:latest" + ports: + - "${DLQ_ADAPTER_PORT:-8089}:8080" + environment: + + cm.amq-adapter.service-name: "amq-dlq" + cm.amq-adapter.is-router: "false" + cm.amq-adapter.generator-id: 165 + cm.amq-adapter.route-mapping: "amq-dlq:DLX:DeadLetterQueue:#:0:0" + + cm.dispatch.use-dlq: false + 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" + quarkus.application.name: AMQ-DeadLetterQueue + quarkus.otel.exporter.otlp.endpoint: "http://jaeger:4317" + OTEL_EXPORTER_OTLP_ENDPOINT: "http://jaeger:4317" + env_file: + - .env_secrets + depends_on: + amqp-router: + condition: service_healthy + # + # test applications + # + amq-clever-book: + image: "amqp_router:latest" + ports: + - "16000:8080" + environment: + + cm.amq-adapter.service-name: "amq-cleverbook-adapter" + cm.amq-adapter.generator-id: 16 + cm.amq-adapter.route-mapping: "amq-clever-book::amq-clever-book:clever.book.localhost.#:5:0" + + 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: clever-book + cm.dispatch.service-port: 8080 + + cm.backpressure.threshold: 20 + cm.backpressure.time-window: 3000 + cm.backpressure.management-service-url: "management-service:8080" + quarkus.otel.service.name: CleverBook-AMQ-Adapter + quarkus.application.name: CleverBook-AMQ-Adapter + quarkus.otel.exporter.otlp.endpoint: "http://jaeger:4317" + OTEL_EXPORTER_OTLP_ENDPOINT: "http://jaeger:4317" + env_file: + - .env_secrets + depends_on: + amqp-router: + condition: service_healthy + + # + # Dummy app allowing POST and GET operations over REST API, emulating a real SpringBoot service + # + clever-book: + build: + context: clever-book/ + dockerfile: Dockerfile + image: "book-management:latest" + ports: + - "16002:8080" + restart: on-failure + environment: + SPRING_PROFILES_ACTIVE: prod + SPRING_APPLICATION_NAME: CleverBook + spring.application.name: CleverBook + OTEL_EXPORTER_OTLP_ENDPOINT: "http://jaeger:4317" + otel.exporter.otlp.endpoint: "http://jaeger:4317" + env_file: + - .env_secrets + depends_on: + amqp-router: + condition: service_healthy + + # + # a black box 3rd party app provisioned as-is + # + black-box-3rd-party: + image: "traefik/whoami" + container_name: "blackbox-third-party" + labels: + - "traefik.enable=true" + - "traefik.http.routers.whoami.rule=Host(`whoami.localhost`)" + - "traefik.http.routers.whoami.entrypoints=web" + + # + # test applications for python + # + amq-clever-pybook: + image: "amq-python:latest" + # image: "amqp_router:latest" + ports: + - "16020:8080" + environment: + + cm.amq-adapter.service-name: "amq-cleverpybook-adapter" + cm.amq-adapter.generator-id: 211 + cm.amq-adapter.route-mapping: "amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0" + + 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: clever-pybook + cm.dispatch.service-port: 5000 + + cm.backpressure.threshold: 20 + cm.backpressure.time-window: 3000 + cm.backpressure.management-service-url: "management-service:8080" + quarkus.otel.service.name: CleverPyBook-AMQ-Adapter + quarkus.application.name: CleverPyBook-AMQ-Adapter + quarkus.otel.exporter.otlp.endpoint: "http://jaeger:4317" + OTEL_EXPORTER_OTLP_ENDPOINT: "http://jaeger:4317" + env_file: + - .env_secrets + depends_on: + amqp-router: + condition: service_healthy + + # + # Dummy app allowing POST and GET operations over REST API, emulating a real SpringBoot service + # + clever-pybook: + image: "clever-pybook:latest" + ports: + - "16012:5000" + restart: on-failure + environment: + OTEL_EXPORTER_OTLP_ENDPOINT: "http://jaeger:4317" + OTEL_EXPORTER_OTLP_INSECURE: "true" + depends_on: + amqp-router: + condition: service_healthy + diff --git a/container-data/alertmanager/alertmanager.yml b/container-data/alertmanager/alertmanager.yml new file mode 100644 index 0000000..5a4a701 --- /dev/null +++ b/container-data/alertmanager/alertmanager.yml @@ -0,0 +1,16 @@ +route: + receiver: 'demo.web.hook' + group_by: ['path', 'queue'] + group_wait: 3s + group_interval: 1s + repeat_interval: 1m +receivers: + - name: 'demo.web.hook' + webhook_configs: + - url: 'http://10.233.1.4:8080/alert' +inhibit_rules: + # override warnings if we have critical + - source_match: + severity: 'critical' + target_match: + severity: 'warning' diff --git a/container-data/grafana/dashboard-export/Application-1730708374720.json b/container-data/grafana/dashboard-export/Application-1730708374720.json new file mode 100644 index 0000000..d58fd5f --- /dev/null +++ b/container-data/grafana/dashboard-export/Application-1730708374720.json @@ -0,0 +1,271 @@ +{ + "__inputs": [ + { + "name": "DS_PROMETHEUS", + "label": "prometheus", + "description": "", + "type": "datasource", + "pluginId": "prometheus", + "pluginName": "Prometheus" + } + ], + "__elements": {}, + "__requires": [ + { + "type": "grafana", + "id": "grafana", + "name": "Grafana", + "version": "11.3.0" + }, + { + "type": "datasource", + "id": "prometheus", + "name": "Prometheus", + "version": "1.0.0" + }, + { + "type": "panel", + "id": "timeseries", + "name": "Time series", + "version": "" + } + ], + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { + "type": "grafana", + "uid": "-- Grafana --" + }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 0, + "id": null, + "links": [], + "panels": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 0 + }, + "id": 2, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "pluginVersion": "11.3.0", + "targets": [ + { + "disableTextWrap": false, + "editorMode": "builder", + "expr": "rabbitmq_queue_messages_ready{queue=\"demo.getRandom\"}", + "fullMetaSearch": false, + "includeNullMetadata": true, + "legendFormat": "__auto", + "range": true, + "refId": "A", + "useBackend": false, + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + } + } + ], + "title": "RabbitMQ Queue getRandom", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 8 + }, + "id": 1, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "pluginVersion": "11.3.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "disableTextWrap": false, + "editorMode": "builder", + "expr": "delta(application_demo_http_req_duration_second_seconds_count[$__interval])", + "fullMetaSearch": false, + "includeNullMetadata": false, + "legendFormat": "__auto", + "range": true, + "refId": "A", + "useBackend": false + } + ], + "title": "/number req", + "type": "timeseries" + } + ], + "schemaVersion": 40, + "tags": [], + "templating": { + "list": [] + }, + "time": { + "from": "now-6h", + "to": "now" + }, + "timepicker": {}, + "timezone": "browser", + "title": "Application", + "uid": "ce2wmjrfkm7lsf", + "version": 5, + "weekStart": "" +} \ No newline at end of file diff --git a/container-data/kibana/config/kibana.yml b/container-data/kibana/config/kibana.yml new file mode 100644 index 0000000..f995ac4 --- /dev/null +++ b/container-data/kibana/config/kibana.yml @@ -0,0 +1,9 @@ +server.host: 0.0.0.0 +elasticsearch.hosts: [ http://elasticsearch:9200 ] + +monitoring.ui.container.elasticsearch.enabled: true +monitoring.ui.container.logstash.enabled: true + +# X-Pack security credentials +elasticsearch.username: kibana_system +elasticsearch.password: ${KIBANA_SYSTEM_PASSWORD} \ No newline at end of file diff --git a/container-data/logstash/pipeline/logstash.conf b/container-data/logstash/pipeline/logstash.conf new file mode 100644 index 0000000..93c3e08 --- /dev/null +++ b/container-data/logstash/pipeline/logstash.conf @@ -0,0 +1,15 @@ +input { + beats { + port => 5044 + } +} + +## Add your filters / logstash plugins configuration here + +output { + elasticsearch { + hosts => "elasticsearch:9200" + user => "logstash_internal" + password => "${LOGSTASH_INTERNAL_PASSWORD}" + } +} diff --git a/container-data/prometheus/alert-rules.yml b/container-data/prometheus/alert-rules.yml new file mode 100644 index 0000000..2f7ea58 --- /dev/null +++ b/container-data/prometheus/alert-rules.yml @@ -0,0 +1,21 @@ +groups: + - name: demo_application + rules: + - alert: application_http_duration_alert + expr: rate(application_demo_http_req_duration_second_seconds_bucket{le="+Inf"}[5s]) >= 100 + for: 5s + labels: + severity: critical + annotations: + summary: "Application process http req take too long" + description: "Endpoint {{ $labels.path }} is taking too long" + - name: rabbitmq + rules: + - alert: rabbitmq_queue_message_ready_count_alert + expr: rabbitmq_queue_messages_ready >= 50 + for: 3s + labels: + severity: critical + annotations: + summary: "RabbitMQ queue backpressure too high" + description: "Queue {{ $labels.queue }} is taking too long" \ No newline at end of file diff --git a/container-data/prometheus/prometheus.yml b/container-data/prometheus/prometheus.yml new file mode 100644 index 0000000..fd35bdb --- /dev/null +++ b/container-data/prometheus/prometheus.yml @@ -0,0 +1,32 @@ +scrape_configs: + - job_name: RabbitMQ + scrape_interval: 2s + metrics_path: /metrics/per-object + docker_sd_configs: + - host: unix:///var/run/docker.sock + relabel_configs: + # Only keep containers that have a `prometheus-job=rabbitmq` label. + # also requires the port = 15962, that's the report port of RabbitMQ + - source_labels: [__meta_docker_container_label_prometheus_job, __meta_docker_port_private] + regex: rabbitmq;15692 + action: keep + + - job_name: SpringActuator + metrics_path: /actuator/prometheus + scrape_interval: 2s + docker_sd_configs: + - host: unix:///var/run/docker.sock + relabel_configs: + - source_labels: [__meta_docker_container_label_prometheus_job, __meta_docker_port_private] + regex: spring-actuator;8080 + action: keep + +alerting: + alertmanagers: + - scheme: http + static_configs: + - targets: + - "alertmanager:9093" + +rule_files: + - "alert-rules.yml" \ No newline at end of file diff --git a/container-data/rabbitmq/enabled_plugins b/container-data/rabbitmq/enabled_plugins new file mode 100644 index 0000000..318ea04 --- /dev/null +++ b/container-data/rabbitmq/enabled_plugins @@ -0,0 +1 @@ +[rabbitmq_management,rabbitmq_prometheus]. \ No newline at end of file diff --git a/container-data/traefik/traefik.yml b/container-data/traefik/traefik.yml new file mode 100644 index 0000000..4c323f6 --- /dev/null +++ b/container-data/traefik/traefik.yml @@ -0,0 +1,20 @@ +api: + insecure: true + dashboard: true + debug: true + disableDashboardAd: true +providers: + docker: + exposedbydefault: false +log: + level: "DEBUG" + +entryPoints: + web: + address: ":8080" + +# ampqs: +# address: ":3333" + +# websecure: +# address: ":443" \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 9049629..9ea4dd5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,8 +1,9 @@ aiohttp~=3.11.11 pika~=1.3.2 -flask opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp opentelemetry-exporter-prometheus +opentelemetry-instrumentation-fastapi opentelemetry-instrumentation-flask +opentelemetry-instrumentation-pika