Compare commits

...

19 Commits

Author SHA1 Message Date
Stanislav Hejny 4d36700887 Make everythong asynch, prepare for streaming uplpaded file 2025-03-13 16:32:11 +00:00
Stanislav Hejny a60c724ae1 Make everythong asynch, prepare for streaming uplpaded file 2025-03-13 16:31:24 +00:00
Stanislav Hejny 5985a22bbb rewrite using asynch PIKA library 2025-03-13 02:13:56 +00:00
Stanislav Hejny a7d7832a49 fix the byte to string conversion 2025-03-02 12:50:46 +00:00
Stanislav Hejny 2e1b95675a fix the byte to string conversion 2025-03-02 12:37:20 +00:00
Stanislav Hejny 1ffd475e52 Fix issues from testing 2025-03-02 00:43:22 +00:00
Stanislav Hejny c4037f3cd9 Fix issues from testing 2025-03-02 00:35:09 +00:00
Stanislav Hejny d9af0d0981 Bump-up project version to enable publish to PyPI 2025-02-28 00:15:00 +00:00
Stanislav Hejny aaa2895cbc remove hardcoded config file path 2025-02-28 00:10:18 +00:00
Stanislav Hejny aaf1c77e72 Fix cleverswarm dependencies identified in testing 2025-02-27 23:08:23 +00:00
Stanislav Hejny dc884c654a Bump-up project version to enable publish to PyPI 2025-02-27 21:50:44 +00:00
Stanislav Hejny e861bf0f38 Remove references to cleverswarm and fix packaging 2025-02-27 21:43:46 +00:00
Stanislav Hejny 30e046b234 Update the Ci/CD pipline to publish AMQ as library 2025-02-27 04:11:05 +00:00
Stanislav Hejny a633ecef4c Update the Ci/CD pipline to publish AMQ as library 2025-02-27 03:56:45 +00:00
Stanislav Hejny 1b0421964c feat/#1 - add the code coverage run to the pipeline 2025-02-26 20:30:08 +00:00
Stanislav Hejny 73c962dc12 Add CI/CD pipeline, adjust for integration with CleverSwarm and other python services 2025-02-26 19:53:44 +00:00
Stanislav Hejny 1f9040729f feat/#1 - refactor code to allow make it as library and to separate the AMQ code from dependencies needed to write the CleverSwarm interface 2025-02-26 10:51:17 +00:00
Stanislav Hejny eb26eec767 CleverSwarm API invocation example 2025-02-25 20:37:33 +00:00
Stanislav Hejny 286a4986b3 AMQ adapter version w/o endpints for CleverSwarm integration 2025-02-20 15:46:53 +08:00
51 changed files with 2219 additions and 730 deletions
+3
View File
@@ -50,3 +50,6 @@ hs_err_pid*
**/__pycache__
.idea
.venv
.coverage
+65
View File
@@ -0,0 +1,65 @@
# This pipeline would run the test coverage report to satisfy Coverage-Check rule.
stages:
- test
- build
- publish
variables:
PROJECT_ID: 217
# Define a cache for pip dependencies to speed up builds
cache:
paths:
- .cache/pip/
# Install dependencies and run tests with coverage
test:
stage: test
image: python:3.11 # Use the desired Python version
before_script:
- python -V # Print Python version for debugging
- pip install --upgrade pip
- pip install -r requirements.txt # Install project dependencies
- pip install pytest coverage # Install testing and coverage tools
script:
- coverage run -m pytest # Run tests with coverage
- coverage report -m # Generate coverage report
- coverage xml # Generate coverage report in XML format for GitLab
artifacts:
reports:
coverage_report:
coverage_format: cobertura
path: coverage.xml
paths:
- coverage.xml
coverage: '/^TOTAL.*\s+(\d+%)$/' # Regex to extract coverage percentage from the report
#rules:
# - if: $CI_COMMIT_BRANCH == "develop" # Run this job only on the 'develop' branch
# Build the package
build:
stage: build
image: python:3.11
before_script:
- pip install --upgrade pip
- pip install setuptools wheel
script:
- python setup.py sdist bdist_wheel
artifacts:
paths:
- dist/
# Publish the package to PyPI
publish:
stage: publish
image: python:3.11
before_script:
- pip install --upgrade pip
- pip install twine
script:
# - twine upload --username $CI_REGISTRY_USER --password $CI_REGISTRY_PASSWORD dist/*
- twine upload --repository-url https://git.cleverthis.com/api/v4/projects/$PROJECT_ID/packages/pypi --username $CI_REGISTRY_USER --password $CI_REGISTRY_PASSWORD dist/*
#only:
# - main # Only publish from the main branch
#rules:
# - if: $CI_COMMIT_BRANCH == "main"
+20
View File
@@ -0,0 +1,20 @@
# 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
# Copy the application code into the container
COPY ./.env_secrets .
COPY ./amqp ./amqp
COPY ./container-data ./container-data
CMD ["python", "amqp/service/amq_service.py"]
+21 -83
View File
@@ -1,107 +1,45 @@
![CleverThis](https://docs.cleverthis.com/public_media/logo-transp.png)
[![pipeline status](https://git.cleverthis.com/cleverthis/base/root/badges/master/pipeline.svg)](https://git.cleverthis.com/cleverthis/base/root/-/commits/master)
[![coverage report](https://git.cleverthis.com/cleverthis/base/root/badges/master/coverage.svg)](https://git.cleverthis.com/cleverthis/base/root/-/commits/master)
[![pipeline status](https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python/badges/master/pipeline.svg)](https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python/-/commits/master)
[![coverage report](https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python/badges/master/coverage.svg)](https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python/-/commits/master)
[![SemVer](https://img.shields.io/badge/SemVer-v2.0.0-green)](https://semver.org/spec/v2.0.0.html)
[![Matrix](https://img.shields.io/matrix/cleverthis%3Aqoto.org?server_fqdn=matrix.qoto.org&label=Matrix%20chat)](https://matrix.to/#/#CleverThis:qoto.org)
# amq-adapter-python
## Getting started
To make it easy for you to get started with GitLab, here's a list of recommended next steps.
Already a pro? Just edit this README.md and make it your own. Want to make it easy? [Use the template at the bottom](#editing-this-readme)!
## Add your files
- [ ] [Create](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#create-a-file) or [upload](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#upload-a-file) files
- [ ] [Add files using the command line](https://docs.gitlab.com/ee/gitlab-basics/add-file.html#add-a-file-using-the-command-line) or push an existing Git repository with the following command:
```
cd existing_repo
git remote add origin https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python.git
git branch -M master
git push -uf origin master
```
## Integrate with your tools
- [ ] [Set up project integrations](https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python/-/settings/integrations)
## Collaborate with your team
- [ ] [Invite team members and collaborators](https://docs.gitlab.com/ee/user/project/members/)
- [ ] [Create a new merge request](https://docs.gitlab.com/ee/user/project/merge_requests/creating_merge_requests.html)
- [ ] [Automatically close issues from merge requests](https://docs.gitlab.com/ee/user/project/issues/managing_issues.html#closing-issues-automatically)
- [ ] [Enable merge request approvals](https://docs.gitlab.com/ee/user/project/merge_requests/approvals/)
- [ ] [Set auto-merge](https://docs.gitlab.com/ee/user/project/merge_requests/merge_when_pipeline_succeeds.html)
## Test and Deploy
Use the built-in continuous integration in GitLab.
- [ ] [Get started with GitLab CI/CD](https://docs.gitlab.com/ee/ci/quick_start/index.html)
- [ ] [Analyze your code for known vulnerabilities with Static Application Security Testing (SAST)](https://docs.gitlab.com/ee/user/application_security/sast/)
- [ ] [Deploy to Kubernetes, Amazon EC2, or Amazon ECS using Auto Deploy](https://docs.gitlab.com/ee/topics/autodevops/requirements.html)
- [ ] [Use pull-based deployments for improved Kubernetes management](https://docs.gitlab.com/ee/user/clusters/agent/)
- [ ] [Set up protected environments](https://docs.gitlab.com/ee/ci/environments/protected_environments.html)
***
# Editing this README
When you're ready to make this README your own, just edit this file and use the handy template below (or feel free to structure it however you want - this is just a starting point!). Thanks to [makeareadme.com](https://www.makeareadme.com/) for this template.
## Suggestions for a good README
Every project is different, so consider which of these sections apply to yours. The sections used in the template are suggestions for most open source projects. Also keep in mind that while a README can be too long and detailed, too long is better than too short. If you think your README is too long, consider utilizing another form of documentation rather than cutting out information.
## Name
Choose a self-explaining name for your project.
## Description
Let people know what your project can do specifically. Provide context and add a link to any reference visitors might be unfamiliar with. A list of Features or a Background subsection can also be added here. If there are alternatives to your project, this is a good place to list differentiating factors.
## Badges
On some READMEs, you may see small images that convey metadata, such as whether or not all the tests are passing for the project. You can use Shields to add some to your README. Many services also have instructions for adding a badge.
## Visuals
Depending on what you are making, it can be a good idea to include screenshots or even a video (you'll frequently see GIFs rather than actual videos). Tools like ttygif can help, but check out Asciinema for a more sophisticated method.
## Installation
Within a particular ecosystem, there may be a common way of installing things, such as using Yarn, NuGet, or Homebrew. However, consider the possibility that whoever is reading your README is a novice and would like more guidance. Listing specific steps helps remove ambiguity and gets people to using your project as quickly as possible. If it only runs in a specific context like a particular programming language version or operating system or has dependencies that have to be installed manually, also add a Requirements subsection.
This project contains the Python implementation of the AMQ adapter
for routing the traffic over the RabbitMQ to intended service and automated service discovery.
## Usage
Use examples liberally, and show the expected output if you can. It's helpful to have inline the smallest example of usage that you can demonstrate, while providing links to more sophisticated examples if they are too long to reasonably include in the README.
## Support
Tell people where they can go to for help. It can be any combination of an issue tracker, a chat room, an email address, etc.
### Install dependencies
## Roadmap
If you have ideas for releases in the future, it is a good idea to list them in the README.
```bash
pip install -r requirements.txt
```
## Contributing
State if you are open to contributions and what your requirements are for accepting them.
### Use as a library
For people who want to make changes to your project, it's helpful to have some documentation on how to get started. Perhaps there is a script that they should run or some environment variables that they need to set. Make these steps explicit. These instructions could also be useful to your future self.
Use this URL to add this project as a dependency:
You can also document commands to lint the code or run tests. These steps help to ensure high code quality and reduce the likelihood that the changes inadvertently break something. Having instructions for running tests is especially helpful if it requires external setup, such as starting a Selenium server for testing in a browser.
```
git+https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python.git@develop#egg=amq_adapter
```
## Authors and acknowledgment
Show your appreciation to those who have contributed to the project.
> Hint: use `git config --global credential.helper store` to enable Git credential manager
> so you don't have to re-enter the gitlab credential everytime.
## License
For open source projects, say how it is licensed.
Then:
## Project status
If you have run out of energy or time for your project, put a note at the top of the README saying that development has slowed down or stopped completely. Someone may choose to fork your project or volunteer to step in as a maintainer or owner, allowing your project to keep going. You can also make an explicit request for maintainers.
```python
from amqp.service import amq_service
# TODO: how to use?
```
## Donating
[![Open Collective backers](https://img.shields.io/opencollective/all/CleverThis)](https://opencollective.com/cleverthis)
As an open-source project we run entierly off donations. Buy one of our hardworking developers a beer by [donating at OpenCollective](https://opencollective.com/cleverthis). All donations go to our bounty fund and allow us to place bounties on important bugs and enhancements. You may also use Beerpay linked via the above badges.
+47 -5
View File
@@ -5,16 +5,15 @@ 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:
SNOWFLAKE_ID_BITS = 80
SEQUENCE_BITS = 12
MACHINE_ID_BITS = 26
MACHINE_ID_BITS = 24
TIMESTAMP_BITS = SNOWFLAKE_ID_BITS - SEQUENCE_BITS - MACHINE_ID_BITS
MACHINE_ID_MASK = (1 << MACHINE_ID_BITS) - 1
@@ -31,11 +30,20 @@ class AMQMessageFactory:
@classmethod
def get_instance(cls, machine_id: int):
"""
return reusable factory instance
:param machine_id: a generator ID, should be unique for each component to prevent duplicate IDs
:return: factory instance
"""
if cls._instance is None:
cls._instance = AMQMessageFactory(machine_id)
return cls._instance
def generate_snowflake_id(self) -> SnowflakeId:
"""
Generate an unique ID following the SnowflakeID structure, see https://en.wikipedia.org/wiki/Snowflake_ID
:return: 80-bit wide ID
"""
timestamp = int(datetime.now(timezone.utc).timestamp() * 1000)
if timestamp < self.snowflake_last_timestamp:
@@ -63,10 +71,19 @@ class AMQMessageFactory:
domain: str,
path: str,
headers: Dict[str, List[str]],
message: str,
message,
) -> AMQMessage:
"""
Encapsulate / hide the system level inputs, create the AMQ message from business relevant input
:param method: a method requested (e.g. HTTP method like GET, POST in case of REST request)
:param domain: service host name
:param path: path at the service
:param headers: HTTP headers
:param message: payload
:return: AMQMessage object
"""
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,
@@ -81,10 +98,20 @@ class AMQMessageFactory:
@staticmethod
def amq_message_routing_key(message: AMQMessage) -> str:
"""
Constructs the message specific routing key.
:param message: AMQMessage input
:return: routing key (as string) compliant to RabbitMQ allowed character set.
"""
return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}")
@staticmethod
def get_timestamp_from_id(_id: SnowflakeId) -> int:
"""
Helper method - retrieve the timestamp from the ID
:param _id: SnowflakeID (includes encoded timestamp)
:return: the message timestamp
"""
bits = _id.get_bits()
low = bits[0] >> (AMQMessageFactory.SEQUENCE_BITS + AMQMessageFactory.MACHINE_ID_BITS)
high = bits[1] << (64 - AMQMessageFactory.SEQUENCE_BITS - AMQMessageFactory.MACHINE_ID_BITS)
@@ -92,6 +119,11 @@ class AMQMessageFactory:
@staticmethod
def to_string(message: AMQMessage) -> str:
"""
Human-readable representation.
:param message: AMQMessage input
:return: as string
"""
return (
f"AMQMessage{{"
f"id={message.id}, "
@@ -106,6 +138,11 @@ class AMQMessageFactory:
@staticmethod
def serialize(message: AMQMessage) -> bytes:
"""
Converts AMQMessage to byte array
:param message: message to serialize
:return: messages as 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="
@@ -123,6 +160,11 @@ class AMQMessageFactory:
@staticmethod
def from_bytes(input_bytes: bytes) -> AMQMessage:
"""
Builds the AMQMessage from its serialized (byte array) form.
:param input_bytes: serialized message
:return: AMQMessage instance
"""
prolog_len = (input_bytes[0] << 8) + input_bytes[1]
metadata = input_bytes[2:prolog_len + 2].decode()
+36
View File
@@ -11,12 +11,43 @@ class AMQResponseFactory:
@staticmethod
def create_async_response_message(_id: SnowflakeId) -> AMQResponse:
"""
Helper method to create standard OK response.
:param _id: ID
:return: Response message
"""
return AMQResponse(
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:
"""
Create the AMQResponse message supplying all key values
:param id: SnowflakeID
:param response_code: HTTP response code
:param content_type: MIME content type
:param body: payload
:param trace_info: Jaeger trace info (id)
:return: AMQResponseMessage object
"""
return 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:
"""
Serialize the message to byte array (to be sent over RabbitMQ)
:param response: Object to serialize
:return: byte array as serialized message
"""
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="
@@ -34,6 +65,11 @@ class AMQResponseFactory:
@staticmethod
def from_bytes(input_bytes: bytes) -> AMQResponse:
"""
Buil;ds the AMQResponse object from its serialized form
:param input_bytes: byte array - serialized response
:return: AMQResponse object
"""
prolog_len = (input_bytes[0] << 8) + input_bytes[1]
metadata = input_bytes[2:prolog_len + 2].decode()
+12 -2
View File
@@ -17,6 +17,11 @@ class AMQRouteFactory:
@staticmethod
def from_string(data):
"""
Builds the AMQRoute from its string form.
:param data: route string
:return: AMQRoute object
"""
try:
keys = data.split(RS)
if len(keys) >= 5:
@@ -33,12 +38,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: "
@@ -55,6 +60,11 @@ class AMQRouteFactory:
@staticmethod
def validate(route_str):
"""
Validates the route string.
:param route_str: route string
:return: boolean validity indicator
"""
if route_str is not None:
return route_str.count(":") > 4
return False
+300
View File
@@ -0,0 +1,300 @@
import asyncio
import json
import logging
import os
import re
import threading
from aiohttp import ClientSession, ClientResponse
from fastapi import HTTPException
from typing import Dict, List, Tuple, Callable, Coroutine, AnyStr, Any
from urllib.parse import parse_qs
from xml.etree import ElementTree
from amqp.adapter.amq_response_factory import AMQResponseFactory
from amqp.service.multipart_parser import CleverMultiPartParser
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import AMQMessage, AMQResponse
# Track the number of messages currently being processed
current_parallel_executions = 0
lock = threading.Lock()
parallel_workers = os.getenv("PARALLEL_WORKERS", default=10)
def parse_request_body(message: AMQMessage):
"""
Parses the HTTP POST request body based on the MIME type.
Args:
message (AMQMessage): Contains raw request and MIME headers.
Returns:
dict: Parsed data as a dictionary.
"""
content_type = message.headers.get('Content-Type', '')
if not content_type:
raise ValueError("Content-Type header is required")
# Parse JSON
if content_type == "application/json":
try:
return json.loads(message.body())
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON: {e}")
# Parse URL-encoded form data
elif content_type == "application/x-www-form-urlencoded":
try:
return {k: v[0] if len(v) == 1 else v for k, v in parse_qs(message.body()).items()}
except Exception as e:
raise ValueError(f"Invalid URL-encoded form data: {e}")
# Parse multipart/form-data
elif content_type.startswith("multipart/form-data"):
try:
parser: CleverMultiPartParser = CleverMultiPartParser(message)
return parser.form_data
except Exception as e:
raise ValueError(f"Invalid multipart/form-data: {e}")
# Parse plain text
elif content_type == "text/plain":
return {"text": message.body()}
# Parse XML
elif content_type == "application/xml":
try:
root = ElementTree.fromstring(message.body())
return {elem.tag: elem.text for elem in root}
except ElementTree.ParseError as e:
raise ValueError(f"Invalid XML: {e}")
else:
raise ValueError(f"Unsupported Content-Type: {content_type}")
# ============= POST Helper function =============
async def call_cleverthis_post_put_api(
message: AMQMessage,
args: Any,
api_method: Callable[[Any, Any], Coroutine],
success_code: int,
authenticated_user: Any
) -> AMQResponse:
"""
Invokes the provided API method with the given arguments and authenticated user,
and returns the response as AMQResponse.
:param message: AMQMessage containing the details of the call and the payload
:param args: map of the arguments to pass to the API method
:param api_method: Callable - the method to invoke. Last parameter is the authenticated user
:param success_code: HTTP code to be returned when operation suceeds
:param authenticated_user: Authenticated user. The actual class depends on the service
:return: AMQResponse class
"""
try:
coroutine_call: Coroutine = api_method(args, authenticated_user)
result: str = await coroutine_call
return AMQResponseFactory.of(
message.id, success_code, message.headers.get("Content-Type", "application/json"),
result.encode('utf-8'), message.trace_info
)
except HTTPException as http_error:
return AMQResponseFactory.of(message.id, http_error.status_code, "text/plain",
str(http_error.detail).encode('utf-8'), message.trace_info)
except Exception as error:
return AMQResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
# ============= GET Helper function =============
async def call_cleverthis_get_api(
message: AMQMessage,
pattern: str,
api_method: Callable[[Tuple[AnyStr | Any, ...], Any], Coroutine],
authenticated_user: Any
) -> AMQResponse:
"""
invokes provided Callable that is expected to be a GET API method, and returns the response as AMQResponse
:param message: AMQMessage containing the details of the call
:param pattern: If there are path variables, use this pattern to extract them
:param api_method: Callable - the method to invoke. Last parameter is the authenticated user
:param authenticated_user: Authenticated user. The actual class depends on the service
:return:
"""
try:
match = re.search(pattern, message.path)
if match:
crt: Coroutine = api_method(match.groups(), authenticated_user)
result: str = await crt
return AMQResponseFactory.of(message.id, 200, "text/plain", result.encode('utf-8'), message.trace_info)
else:
_msg = f"Unexpected URL: {message.path}, expected pattern: {pattern}"
return AMQResponseFactory.of(message.id, 404, "text/plain", _msg.encode('utf-8'), message.trace_info)
except HTTPException as http_error:
return AMQResponseFactory.of(message.id, http_error.status_code, "text/plain",
str(http_error.detail).encode('utf-8'), message.trace_info)
except Exception as error:
return AMQResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
# ============= CleverThisServiceAdapter =============
class CleverThisServiceAdapter:
"""
A base class for CleverThis service adapter that handles the incoming AMQP messages and routes them to the
appropriate CleverThisService API endpoint.
"""
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
async def get_current_user(self, token: str):
"""
To Be Overriden in subclass for given CleverThis Service, to return the current user based on the token,
in format appropriate for the service.
:return: A user object or None if the user is not authenticated.
"""
return None
def sync_on_message(self, message: AMQMessage) -> AMQResponse:
"""
Synchronous version of on_message method. This is a wrapper around the async on_message method.
:param message: AMQMessage
:return: AMQResponse
"""
return asyncio.run(self.on_message(message))
async def on_message(self, message) -> AMQResponse:
"""
Called when a message is received from the AMQP.
:param message:
:return: AMQP response class with operation status code and optional error details.
"""
DEBUG_NO_AUTH = 1
global current_parallel_executions
# Increment the counter for parallel executions
with lock:
current_parallel_executions += 1
if current_parallel_executions > parallel_workers - 2:
logging.warn("Warning: CApacity close to depleted!")
_auth: str = message.headers.get('Authorization', '')
_auth_user: Any | None = None
if 'Bearer' in _auth:
try:
token = _auth.split(' ')[1]
_auth_user = await self.get_current_user(token)
except Exception as error:
logging.error(f"Error getting user: {error}")
amq_response: AMQResponse | None = None
if _auth_user or DEBUG_NO_AUTH == 1:
method = message.method.lower()
if method == 'get':
amq_response = await self.get(message, _auth_user)
elif method == 'put':
amq_response = await self.put(message, _auth_user)
elif method == 'post':
amq_response = await self.post(message, _auth_user)
elif method == 'head':
amq_response = await self.head(message, _auth_user)
elif method == 'delete':
amq_response = await self.delete(message, _auth_user)
else:
raise ValueError(f"Unexpected HTTP method: {message.method}")
else:
amq_response = AMQResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info)
with lock:
current_parallel_executions -= 1
return amq_response
# ==================================================================================================
# ===================== C l e v e r T h i s A P I m e t h o d s ============================
# ==================================================================================================
async def get(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
"""
Handles the GET requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
:param message: message from the AMQP that contains the GET request
:param auth_user: user authenticated by the API Gateway, as UserSchem
:return: AMQResponse to be sent back to the API Gateway via AMQP
"""
# unmatched path - try to invoke the service directly on this path
url = f"http://{self.service_host}:{self.service_port}{message.path}"
headers = {**message.headers, **message.trace_info}
logging.info(f" [*] SEND HEADERS out: {', '.join(f'{k}={v}' for k, v in headers.items())}, REQ={url}")
if self.session:
_http_resp: ClientResponse = await self.session.get(url, headers=headers)
return AMQResponseFactory.of(message.id, _http_resp.status, _http_resp.content_type,
await _http_resp.read(),
message.trace_info)
return AMQResponseFactory.of(message.id, 404, "text/plain",
str("Destination location does not exists").encode('utf-8'), message.trace_info)
async def put(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
"""
Handles the PUT requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
:param message: message from the AMQP that contains the PUT request
:param auth_user: user authenticated by the API Gateway, as UserSchem
:return: AMQResponse to be sent back to the API Gateway via AMQP
"""
return 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},
),
auth_user
)
async def post(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
"""
Handles the POST requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
:param message: message from the AMQP that contains the POST request, including body and headers
:param auth_user: user authenticated by the API Gateway, as UserSchem
:return: AMQResponse to be sent back to the API Gateway via AMQP
"""
return await self.handle_possible_form(
message,
self.session.post(
f"http://{self.service_host}:{self.service_port}{message.path}",
data=message.body,
headers={**message.headers, **message.trace_info},
),
auth_user
)
async def handle_possible_form(self, message: AMQMessage, request_coroutine, auth_user: Any) -> AMQResponse:
args = parse_request_body(message)
return await request_coroutine
async def head(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
"""
Handles the HEAD requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
:param message: message from the AMQP that contains the POST request, including body and headers
:param auth_user: user authenticated by the API Gateway, as UserSchem
:return: AMQResponse to be sent back to the API Gateway via AMQP
"""
raise NotImplementedError
async def delete(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
"""
Handles the DELETE requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
:param message: message from the AMQP that contains the POST request, including body and headers
:param auth_user: user authenticated by the API Gateway, as UserSchem
:return: AMQResponse to be sent back to the API Gateway via AMQP
"""
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'
-68
View File
@@ -1,68 +0,0 @@
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'
-15
View File
@@ -1,15 +0,0 @@
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
+37 -5
View File
@@ -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
@@ -15,11 +16,18 @@ INVALID_MESSAGE = CleverMicroServiceMessage() # default values mean invalid mes
class CleverMicroServiceMessageFactory:
"""
build/serialize/deserialize CleverMicro Service Message class
"""
def __init__(self, name: str):
self.process_name = name
def to_string(self, message: CleverMicroServiceMessage) -> str:
"""
Convert CleverMicroServiceMessage to human-readable format
:param message: CleverMicroServiceMessage object
:return: as string
"""
if message is not None:
return (
f"CleverMicroServiceMessage{{id={message.id.as_string()}|type={self.format_message_type(message)}|"
@@ -29,6 +37,13 @@ class CleverMicroServiceMessageFactory:
return "null"
def of(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> CleverMicroServiceMessage:
"""
Builds CleverMicroServiceMessage object from relevant values.
:param message_type: enum Message type
:param payload: Message content, stored as-is
:param recipient_name: Recipient name (from the routing expression)
:return: CleverMicroServiceMessage object
"""
_id = AMQMessageFactory.get_instance(1).generate_snowflake_id()
return CleverMicroServiceMessage(
id=_id,
@@ -41,6 +56,13 @@ class CleverMicroServiceMessageFactory:
)
def as_base64(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> CleverMicroServiceMessage:
"""
Builds CleverMicroServiceMessage object from relevant values, stores content Base64 encoded.
:param message_type: enum Message type
:param payload: Message content, stored as Base64 string
:param recipient_name: Recipient name (from the routing expression)
:return: CleverMicroServiceMessage object
"""
_id = AMQMessageFactory.get_instance(1).generate_snowflake_id()
return CleverMicroServiceMessage(
id=_id,
@@ -53,7 +75,12 @@ class CleverMicroServiceMessageFactory:
)
def from_bytes(self, input_bytes: bytes) -> CleverMicroServiceMessage:
print(f"CleverMicroServiceMessage >>> [{input_bytes.decode()}]")
"""
Deserialize CleverMicroServiceMessage from byte array serialized form.
:param input_bytes: serialized CleverMicroServiceMessage
:return: CleverMicroServiceMessage object represented by the input.
"""
logging.info(f"CleverMicroServiceMessage >>> [{input_bytes.decode()}]")
input_str_array = input_bytes.decode().split(RS)
try:
i = 0
@@ -83,11 +110,16 @@ 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
def serialize(self, message: CleverMicroServiceMessage) -> bytes:
"""
Serialize CleverMicroServiceMessage to byte array.
:param message: CleverMicroServiceMessage to serialize.
:return: serialized object.
"""
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}"
@@ -97,7 +129,7 @@ class CleverMicroServiceMessageFactory:
def format_message_type(self, message: CleverMicroServiceMessage) -> str:
"""
Formats the Enum message type and ecorates it with base64 flag at highest bit.
Formats the Enum message type and decorates it with base64 flag at highest bit.
For compatibility with 0-based enum ordinals in Java subtract 1 from ordinal value.
:param message: Service message to send
:return: formatted message type
+7 -7
View File
@@ -9,7 +9,7 @@ class AMQMessageFactoryTest(TestCase):
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.id.as_string(),'64F3B5AF4D00000A4000'.lower(),'SnowflakeID incorrectly parsed.')
self.assertEqual(msg.headers["Content-Type"][0], "multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v")
def test_generate_snowflake_id(self):
@@ -24,10 +24,10 @@ class AMQMessageFactoryTest(TestCase):
_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')
self.assertEqual('0007F000'.lower(), _i0.as_string()[-8:], 'Instance ID and sequence not set correctly')
self.assertEqual('0007F001'.lower(), _i1.as_string()[-8:], 'Instance ID and sequence not set correctly')
self.assertEqual('0007F002'.lower(), _i2.as_string()[-8:], 'Instance ID and sequence not set correctly')
self.assertEqual('0007F003'.lower(), _i3.as_string()[-8:], 'Instance ID and sequence not set correctly')
def test_create_request_message(self):
_f = AMQMessageFactory(34)
@@ -53,7 +53,7 @@ class AMQMessageFactoryTest(TestCase):
_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")
self.assertEqual("localhost.test.me", _key, "Message routing key wrongly extracted from the message")
def test_get_timestamp_from_id(self):
_f = AMQMessageFactory(35)
@@ -65,4 +65,4 @@ class AMQMessageFactoryTest(TestCase):
_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")
self.assertGreater(_as_str.index("domain='localhost'"), 0, "Expected content not found")
+12 -7
View File
@@ -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.adapter.service_message_factory import CleverMicroServiceMessageFactory, RS, INVALID_MESSAGE
from amqp.model.model import CleverMicroServiceMessage
from amqp.model.service_message_type import ServiceMessageType
class CleverMicroServiceMessageTest(TestCase):
@@ -10,13 +11,15 @@ class CleverMicroServiceMessageTest(TestCase):
_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
self.assertTrue("|type=01|" in message_str)
self.assertTrue("|body=Duck" in message_str)
message: CleverMicroServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald")
message_str = _f.to_string(message)
assert "|type=03|" in message_str
assert "|body=Duck" in message_str
assert "|replyTo=bumble-bee" in message_str
self.assertTrue("|type=03|" in message_str)
self.assertTrue("|body=Duck" in message_str)
self.assertTrue("|replyTo=bumble-bee" in message_str)
nullMsg = _f.to_string(None)
self.assertEqual("null", nullMsg)
def test_from(self):
_f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router")
@@ -28,6 +31,8 @@ class CleverMicroServiceMessageTest(TestCase):
assert deserialized.recipient_name == "Donald"
assert deserialized.message == "Duck"
assert deserialized.reply_to == "amqp-router"
invalidMsg = _f.from_bytes(b'wrong')
self.assertEqual(invalidMsg, INVALID_MESSAGE)
def test_base64(self):
_f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router")
+1 -1
View File
@@ -4,7 +4,7 @@ from typing import Dict
@dataclass
class TraceInfoAdapter:
"""
This class should set up the Open Telemetry tracing stack. To be done later.
TODO This class should set up the Open Telemetry tracing stack. To be done later.
"""
@staticmethod
+35 -8
View File
@@ -1,8 +1,15 @@
import configparser
import os
"""
Convert configuration from properties file to an object.
"""
class AMQAdapter:
"""
configuration with prefix 'cm.amq-adapter'
"""
PREFIX: str = 'cm.amq-adapter.'
@@ -15,15 +22,18 @@ 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:
return f"{self.service_name}-{self.generator_id}-"
class Dispatch:
"""
configuration with prefix 'cm.dispatch'
"""
PREFIX: str = 'cm.dispatch.'
@@ -44,13 +54,16 @@ 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:
"""
configuration with prefix 'cm.backpressure'
"""
PREFIX: str = 'cm.backpressure.'
@@ -68,11 +81,25 @@ class Backpressure:
class AMQConfiguration:
def __init__(self):
"""
Top level configuration context holder / object
"""
def __init__(self, config_file_name: str = None):
config = configparser.ConfigParser()
# Read the application.properties file
config.read('config/application.properties')
if os.path.exists(config_file_name):
config.read(config_file_name)
else:
print(f" [E] Configuration file not found: {config_file_name}")
print(" [W] Defaulting to preset values, which will likely be wrong, causing errors.")
# Override values with environment variables
for section in config.sections():
for option in config.options(section):
# env_var = f"{section.upper()}_{option.upper()}"
if option in os.environ:
config.set(section, option, os.environ[option])
#
# Add individual config areas
#
+1 -1
View File
@@ -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-cleverswarm::amq-cleverswarm:cleverswarm.localhost.#:5:0
cm.dispatch.use-dlq=true
cm.dispatch.use-confirms=false
+18
View File
@@ -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-cleverswarm::amq-cleverswarm:cleverswarm.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
+11
View File
@@ -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)
+38 -23
View File
@@ -1,53 +1,62 @@
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
RS = ":"
"""
Define the structure of messages used in Clever Microś
"""
@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 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}"
"""
Build the unique identifier of the route
:return primary key
"""
return f"{self.component_name}{RS}{self.queue}{RS}{self.key}"
def as_string(self) -> str:
return self.__str__()
@@ -55,13 +64,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 +88,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
+10
View File
@@ -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()
+22 -18
View File
@@ -2,25 +2,28 @@ import struct
class SnowflakeId:
"""
see https://en.wikipedia.org/wiki/Snowflake_ID
"""
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):
def __init__(self, hi_bits_or_bytes = None, lo_bits = 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]
if lo_bits is None:
if hi_bits_or_bytes is not None:
i = 0
shift = 8 # starting from Hi value, which is only 2 bytes wide
n = 1
while i < len(hi_bits_or_bytes) and i <= self.SNOWFLAKE_ID_WIDTH:
self.bits[n] |= hi_bits_or_bytes[i] << shift
if shift == 0:
n -= 1
shift = 56
else:
shift -= 8
i += 1
else:
self.bits = [lo_bits, hi_bits_or_bytes]
@staticmethod
def from_hex(_input):
@@ -42,7 +45,8 @@ class SnowflakeId:
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):]
_val = _hex if len(_hex) <= 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH else _hex[(-2 * SnowflakeId.SNOWFLAKE_ID_WIDTH):]
return _val.lower()
def __eq__(self, other):
if isinstance(other, SnowflakeId):
@@ -59,7 +63,7 @@ class SnowflakeId:
def get_bytes(self):
high = struct.pack("!Q", self.bits[1])
low = struct.pack("!Q", self.bits[0])
return high[2:] + low
return high[-2:] + low
def get_bits(self):
return self.bits
+48
View File
@@ -0,0 +1,48 @@
from unittest import TestCase
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.model.snowflake_id import SnowflakeId
class SnowflakeIdTest(TestCase):
def test_equals(self):
_id_1 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
_id_2 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
self.assertFalse(_id_1 == _id_2)
_id_3 = SnowflakeId(_id_1.get_bytes())
self.assertTrue(_id_1 == _id_3)
def test_lt(self):
_id_1 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
_id_2 = AMQMessageFactory.get_instance(1).generate_snowflake_id()
self.assertTrue(_id_1 < _id_2)
def test_from_hex(self):
_factory = AMQMessageFactory.get_instance(1)
_id_1 = _factory.generate_snowflake_id()
_id_1_ser = _id_1.as_string()
_id_2 = SnowflakeId.from_hex(_id_1_ser)
self.assertTrue(_id_1 == _id_2)
with self.assertRaises(RuntimeError) as context:
SnowflakeId.from_hex("100")
self.assertTrue(f"MessageId is {SnowflakeId.SNOWFLAKE_ID_WIDTH * 8} bits wide, it is represented by HEX string "
in str(context.exception))
def test_get_bytes(self):
_factory = AMQMessageFactory.get_instance(1)
_id_1 = _factory.generate_snowflake_id()
_id_1_ser = _id_1.get_bytes()
self.assertEqual(SnowflakeId.SNOWFLAKE_ID_WIDTH, len(_id_1_ser))
def test_get_bits(self):
_factory = AMQMessageFactory.get_instance(1)
_id_1 = _factory.generate_snowflake_id()
self.assertEqual(2, len(_id_1.get_bits()))
def test_as_string(self):
_factory = AMQMessageFactory.get_instance(1)
_id_1 = _factory.generate_snowflake_id()
self.assertEqual(2 * SnowflakeId.SNOWFLAKE_ID_WIDTH, len(_id_1.as_string()))
+21
View File
@@ -0,0 +1,21 @@
import io
class UploadedFile:
def __init__(self, filename: str = None):
self.filename = filename
self.body_b64 = io.BytesIO()
def from_bytes(input_bytes: bytes) -> UploadedFile:
"""
Builds the UploadedFile from its serialized (byte array) form.
:param input_bytes: serialized message
:return: UploadedFile instance
"""
prolog_len = (input_bytes[0] << 8) + input_bytes[1]
metadata = input_bytes[2:prolog_len + 2].decode()
body_b64 = input_bytes[prolog_len + 2:]
return UploadedFile()
+45 -108
View File
@@ -1,7 +1,8 @@
import logging
import pika
from pika.exchange_type import ExchangeType
from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange
from pika.exchange_type import ExchangeType
from amqp.model.model import AMQRoute
from amqp.rabbitmq.rabbit_mq_producer import RabbitMQProducer
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE, DLQ_EXCHANGE, AMQ_RPC_EXCHANGE
@@ -15,6 +16,20 @@ class RabbitMQConsumer(RabbitMQProducer):
def __init__(self, configuration):
super().__init__(configuration, RouterConsumer(configuration))
self.reply_to_exchange: AbstractRobustExchange | None = None # when sending response to RPC request
self.reply_to_queue: AbstractRobustQueue | None= None # when receiving response to RPC request
async def init(self, loop) -> AbstractRobustExchange:
await self.setup_amq_channel(loop)
cm_reply_to_queue = self.router.get_reply_to_queue_name()
self.reply_to_queue = await self.channel.declare_queue(
name=cm_reply_to_queue, durable=True, exclusive=False, auto_delete=False
)
self.reply_to_exchange = await self.channel.declare_exchange(
AMQ_REPLY_TO_EXCHANGE, type=pika.exchange_type.ExchangeType.direct
)
await self.reply_to_queue.bind(exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue)
return self.reply_to_exchange
def cleanup(self):
logging.info(" [*] RabbitMQConsumer.cleanup()")
@@ -22,122 +37,44 @@ class RabbitMQConsumer(RabbitMQProducer):
if self.router.is_open(self.channel):
try:
self.channel.close()
except (pika.connection.exceptions.AMQPError, pika.connection.exceptions.ChannelClosedByBroker) as e:
except Exception as e:
logging.error(" [E] RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e)
def register_routes(self):
async def register_inbound_routes(self, route_mapping: str, callback):
if self.router.is_open(self.channel):
queue_args = self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None
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)
queue: AbstractRobustQueue = await self.channel.declare_queue(name=queue_name,
durable=True,
exclusive=False,
auto_delete=False,
arguments=queue_args)
exchange_name: str = route.exchange if route.exchange else AMQ_RPC_EXCHANGE
exchange: AbstractRobustExchange = await self.channel.declare_exchange(
name=exchange_name, type=ExchangeType.topic
)
)
logging.info(" [*] AMQ connection initialized for route: %s, listens on: %s/%s",
route, exchange, queue_name)
await queue.bind(exchange, route.key)
await queue.consume(callback, no_ack=False)
await self.router.register_route(
AMQRoute(
route.component_name, exchange_name, queue_name,
sanitize_as_rabbitmq_name(route.key),
route.timeout, route.valid_until
)
)
logging.info(" [*] AMQ connection initialized for route: %s, listens on: %s/%s",
route, exchange, queue_name)
except Exception as err:
logging.error(" [E] Callback registration INCOMPLETE, %s", err)
else:
logging.warning(" [W] Channel is null, cannot register routes.")
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
async def register_rpc_handler(self, rpc_consumer):
await self.reply_to_queue.consume(callback=rpc_consumer,
no_ack=False
)
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)
"""
+44 -74
View File
@@ -1,7 +1,5 @@
import logging
import time
import pika
import aio_pika
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
@@ -11,38 +9,22 @@ 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.channel: aio_pika.RobustChannel | None = None
self.connection: aio_pika.RobustConnection | None = None
self.rpc_exchange: aio_pika.RobustExchange | None = None
self.reply_exchange: aio_pika.RobustExchange | None = None
self.amq_configuration = amq_configuration
logging.info("*******************************************")
logging.info("******** RabbitMQProducer.init() ********** %s - %s - %s",
self.amq_configuration.dispatch.amq_host,
self.amq_configuration.dispatch.use_confirms,
self.WINDOW)
logging.info("******** RabbitMQProducer.init(): %s",
self.amq_configuration.dispatch.amq_host)
logging.info("*******************************************")
self.router = router or RouterProducer(amq_configuration)
# this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
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
async def init(self, loop):
await self.setup_amq_channel(loop)
def cleanup(self):
if self.connection:
@@ -51,27 +33,27 @@ class RabbitMQProducer:
except Exception as e:
logging.error("RabbitMQConsumer.PreDestroy cleanup - connection close error: %s", e)
def setup_amq_channel(self):
async def setup_amq_channel(self, loop):
"""
Set up the RabbitMQ connection
:return: nothing, it applies changes to instance variables
"""
factory = pika.ConnectionParameters(
self.connection = await aio_pika.connect_robust(
host=self.amq_configuration.dispatch.amq_host,
username=self.amq_configuration.dispatch.rabbit_mq_user,
password=self.amq_configuration.dispatch.rabbit_mq_password
port=self.amq_configuration.dispatch.amq_port,
login=self.amq_configuration.dispatch.rabbit_mq_user,
password=self.amq_configuration.dispatch.rabbit_mq_password,
loop=loop
)
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)
logging.info("RabbitMQProducer.setupAMQ, got connection, open=%s", self.connection.is_closed)
self.channel = await self.connection.channel()
await self.channel.set_qos(prefetch_count=self.PREFETCH_COUNT)
await self.router.init_routing_paths(self.channel)
self.router.set_route_added_notifier(
lambda route: logging.info(" [DBG] ---------- RouteAdded, setting up metrics: amqp.adapter.exchange.%s",
route.exchange))
def send_message(self, message: AMQMessage):
async def send_message(self, message: AMQMessage):
"""
The API routine that sends a datagram message (AMQMessage) to the destination. The Destination
is determined from the AMQMessage metadata section, using `domain` and `path` as key.
@@ -81,54 +63,42 @@ class RabbitMQProducer:
"""
try:
route = self.router.find_route_by_message(message)
if route:
if route.timeout < 0:
if route and self.rpc_exchange:
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)
pika_message: aio_pika.message.Message = aio_pika.message.Message(
body=AMQMessageFactory.serialize(message),
content_encoding='utf-8',
delivery_mode=2
)
await self.rpc_exchange.publish(
message=pika_message,
routing_key=self.router.get_routing_key(message), # context path is a routing key
)
logging.info(f" [x] basicPublish (no-reply), messageId: {message.id}, to: {route}")
else:
# This is RPC call, there should be response
props = pika.BasicProperties(
correlation_id=message.id,
pika_message: aio_pika.message.Message = aio_pika.message.Message(
body=AMQMessageFactory.serialize(message),
content_encoding='utf-8',
delivery_mode=2,
correlation_id=str(message.id),
reply_to=self.router.get_reply_to_queue_name()
)
self.channel.basic_publish(
route.exchange,
self.router.get_routing_key(message), # context path is a routing key
properties=props,
body=AMQMessageFactory.serialize(message)
await self.rpc_exchange.publish(
message=pika_message,
routing_key=self.router.get_routing_key(message), # context path is a routing key
)
logging.info(" [x] basicPublish (RPC call), messageId: %s, to: %s",
message.id.as_string(), route.as_string())
else:
logging.warning("Message not sent, no route for {}/{}", 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)
def request_routes(self):
self.router.request_routes_from_adapters()
def setup_local_ack(self):
self.channel.confirm_delivery()
async def request_routes(self):
await self.router.request_routes_from_adapters()
def find_route(self, domain: str, path: str) -> AMQRoute:
return self.router.find_route(domain, path)
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")
+24 -8
View File
@@ -14,14 +14,30 @@ class RouteDatabase:
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"\."
breaker = False
tokens = routing_key.split('.')
counter = len(tokens)
regex_parts = []
for token in tokens:
counter -= 1
if not breaker:
if token == "":
res = "[^\\.]*"
elif token == "#":
breaker = True
res = ".*$"
else:
res = token
if counter > 0 and not breaker:
res += "\\."
regex_parts.append(res)
else:
regex_parts.append("")
regex = "".join(filter(lambda x: x != "", regex_parts))
if not routing_key:
return re.compile(".*")
return re.compile(regex)
def matches(self, routing_key_from_route: str, routing_key_from_message: str) -> bool:
+2 -2
View File
@@ -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)
+28 -35
View File
@@ -1,12 +1,11 @@
import logging
from typing import Set
from pika import spec
from pika.adapters.blocking_connection import BlockingChannel
from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractIncomingMessage
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import CleverMicroServiceMessage, 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,25 +28,23 @@ class RouterConsumer(RouterProducer):
self.amqRouteFactory = AMQRouteFactory()
logging.info(" [*] RouterConsumer.<init>")
def init_routing_paths_consumer(self, _channel: BlockingChannel):
async def init_routing_paths_consumer(self, _channel: AbstractRobustChannel):
"""
* Initialize router according to Consumer mode.
"""
self.init_routing_paths(_channel)
logging.info(" [*] RouterConsumer.initConsumerRouter, channel open::", self.is_open(_channel))
await 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
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)
queue: AbstractRobustQueue = await self.channel.declare_queue(name=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="")
await queue.bind(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)
logging.info(" [*] RouterConsumer listens for RouteDataRequests on: %s", cm_request_queue)
await queue.consume(callback=self.handle_routing_data_request_message, no_ack=False)
except Exception as ioe:
logging.error(" [E] RouterConsumer FATAL: cannot initialize, %s", ioe)
else:
@@ -57,39 +54,35 @@ class RouterConsumer(RouterProducer):
* 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):
async def handle_routing_data_request_message(self, message: AbstractIncomingMessage):
# some debug statements, TODO - remove for prod release
delivery_tag = method.delivery_tag
debug = "Delivery.Properties = " + properties.__str__()
delivery_tag = message.delivery_tag
debug = "Delivery.Properties = " + message.properties.__str__()
logging.info(" [DBG] ******[%] (ROUTING_DATA_REQ) tag: %s, env: %s, props: %s",
delivery_tag, method.consumer_tag, method.get_properties(), debug)
delivery_tag, message.consumer_tag, message.properties, debug)
self.channel.basic_ack(delivery_tag, False)
await message.ack(False)
message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(body)
cm_message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(message.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)
logging.info(" [DBG] ******[%s] (ROUTING_DATA_REQ) msg=%s, routeMap=[%s]",
delivery_tag, self.service_message_factory.to_string(cm_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:
if cm_message.is_valid() and cm_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
ServiceMessageType.ROUTING_DATA, route_mapping, cm_message.reply_to
)
self.publish_service_message(service_message, CM_RESPONSE_EXCHANGE)
await self.publish_service_message(service_message, self.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))
delivery_tag, str(message.body))
def register_route(self, route: AMQRoute):
async 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.
@@ -100,29 +93,29 @@ class RouterConsumer(RouterProducer):
serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA, route.as_string(), ANY_RECIPIENT
)
self.publish_service_message(serviceMessage, CM_RESPONSE_EXCHANGE)
await self.publish_service_message(serviceMessage, self.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)
except Exception as ioe:
logging.error(" [E] RouterConsumer failed register route data: %s", ioe)
def remove_route_on_cleanup(self, route: AMQRoute) -> bool:
async 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())
logging.info(" [DBG] RouterConsumer.removeRoute, route: %s", 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):
if not await self.publish_service_message(service_message, self.cm_response_exchange):
logging.warning(" [W] RouterConsumer couldn't remove current route: %s, channel not open.",
route)
+55 -64
View File
@@ -7,15 +7,16 @@
* forwards it to the application.
"""
import logging
import os
import pika
from pika import BasicProperties, spec
from pika.adapters.blocking_connection import BlockingChannel
from aio_pika import Message
from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractRobustConnection, AbstractRobustExchange, \
AbstractIncomingMessage
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
@@ -28,26 +29,18 @@ class RouterProducer(RouterBase):
* :param configuration: adapter's configuration reference.
"""
super().__init__()
self.connection = None
self.connection: AbstractRobustConnection | None = 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: AbstractRobustChannel | None = None
self.cm_request_exchange: AbstractRobustExchange | None = None
self.cm_response_exchange: AbstractRobustExchange | None = None
logging.info(f" [*] RouterProducer.<init>: %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):
async def init_routing_paths(self, channel: AbstractRobustChannel | None):
"""
* Initialize router according to Producer mode.
* The logic here sets a producer and listener on CleverMicroRequest/CleverMicroResponse queues
@@ -62,23 +55,25 @@ class RouterProducer(RouterBase):
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)
self.cm_request_exchange = await self.channel.declare_exchange(
name=CM_REQUEST_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)
self.cm_response_exchange = await self.channel.declare_exchange(
name=CM_RESPONSE_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={})
response_queue: AbstractRobustQueue = await self.channel.declare_queue(
name=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="#")
await response_queue.bind(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()
await response_queue.consume(callback=self.handle_routing_data_message, no_ack=False)
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)
@@ -86,38 +81,34 @@ class RouterProducer(RouterBase):
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):
async def handle_routing_data_message(self, message: AbstractIncomingMessage):
"""
* 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 {message.body}, m={message.message_id}, p={message.properties}")
"""
logging.info(" [DBG] ******[%s] ROUTING_DATA %s >>> %s",
delivery.getEnvelope().getDeliveryTag(),
delivery.getEnvelope().toString(), str(delivery.getBody()))
"""
# 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)
await message.ack(multiple=False)
cm_message: CleverMicroServiceMessage = self.service_message_factory.from_bytes(message.body)
# ignore the message if it comes from ourselves (even if from different instance)
logging.info(" [DBG] ******[%s] ROUTING_DATA, msg=%s, routeMap=[{] sn={, replyTo={",
method.delivery_tag,
self.service_message_factory.to_string(message),
logging.info(" [DBG] ******[%s] ROUTING_DATA, msg=%s, routeMap=[%s] sn=%s, replyTo=%s",
message.delivery_tag,
self.service_message_factory.to_string(cm_message),
self.amq_configuration.amq_adapter.route_mapping,
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)
cm_message.reply_to)
if cm_message.is_valid() and not self.amq_configuration.amq_adapter.service_name == cm_message.reply_to:
if cm_message.message_type == ServiceMessageType.ROUTING_DATA:
self.add_routes(cm_message.message)
if cm_message.message_type == ServiceMessageType.ROUTING_DATA_REMOVE:
self.remove_routes(cm_message.message)
def request_routes_from_adapters(self):
async def request_routes_from_adapters(self):
"""
* Send a broadcast message prompting all listening adapters to reply with own routing data.
"""
@@ -126,37 +117,37 @@ class RouterProducer(RouterBase):
serviceMessage: CleverMicroServiceMessage = self.service_message_factory.of(
ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT
)
if not self.publish_service_message(serviceMessage, CM_REQUEST_EXCHANGE):
if not await self.publish_service_message(serviceMessage, self.cm_request_exchange):
logging.warning(" [E] RouterProducer could not request route data: %s channel is not open.",
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:
async def publish_service_message(self,
message: CleverMicroServiceMessage,
exchange: AbstractRobustExchange) -> 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)
* :param exchange: target exchange name where to publish the message
* :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))
pika_message: Message = Message(
body=binary_content,
content_encoding='utf-8',
delivery_mode=2,
content_type="application/octet-stream",
headers=None,
priority=0,
correlation_id=None
)
await exchange.publish(message=pika_message, routing_key="")
logging.info(" [x] Service Message Sent to %s, msg: %s", exchange.name, str(binary_content))
return True
return False
@@ -181,10 +172,10 @@ 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: AbstractRobustChannel) -> 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
return _channel is not None and not _channel.is_closed
+86 -7
View File
@@ -1,24 +1,103 @@
from unittest import TestCase
import pytest
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.model.model import AMQRoute
from amqp.router.route_database import RouteDatabase
ROUTE_1 = "amq::amq1:clever1.book.localhost.#:5:0"
ROUTE_2 = "amq::amq2:clever2.book.localhost.#:5:0"
ROUTE_3 = "amq::amq3:clever3.book.localhost.#:5:0"
ROUTE_7 = "amq::amq7:clever7.book.localhost.#:5:0"
class TestRouteDatabase(TestCase):
@pytest.fixture(autouse=True)
def setup_each_test(self):
self._database = RouteDatabase()
if len(self._database.get_routes()) == 0:
r1 = AMQRouteFactory.from_string(ROUTE_1)
r2 = AMQRouteFactory.from_string(ROUTE_2)
r3 = AMQRouteFactory.from_string(ROUTE_3)
self._database.add_route(r1)
self._database.add_route(r2)
self._database.add_route(r3)
def test_get_routes(self):
self.fail()
_amq_factory = AMQRouteFactory()
_route: AMQRoute = _amq_factory.from_string("amq-clever-pybook::clever-pybook:clever.pybook.localhost.#:5:0")
_routes = self._database.get_routes()
self.assertIsNotNone(_routes)
self.assertEqual(3, len(_routes))
self._database.add_route(_route)
_routes = self._database.get_routes()
self.assertTrue(_route in _routes, "Route was not added")
self.assertEqual(4, len(_routes))
def test_pattern(self):
self.fail()
pattern = self._database.pattern("")
self.assertEqual(".*", pattern.pattern)
pattern = self._database.pattern("..")
self.assertEqual("[^\\.]*\\.[^\\.]*\\.[^\\.]*", pattern.pattern)
pattern = self._database.pattern(".#.#")
self.assertEqual("[^\\.]*\\..*$", pattern.pattern)
pattern = self._database.pattern("..#")
self.assertEqual("[^\\.]*\\.[^\\.]*\\..*$", pattern.pattern)
pattern = self._database.pattern("#")
self.assertEqual(".*$", pattern.pattern)
pattern = self._database.pattern("ba.dc.")
self.assertEqual("ba\\.dc\\.[^\\.]*", pattern.pattern)
pattern = self._database.pattern(".cd.")
self.assertEqual("[^\\.]*\\.cd\\.[^\\.]*", pattern.pattern)
pattern = self._database.pattern("de")
self.assertEqual("de", pattern.pattern)
def test_matches(self):
self.fail()
self.assertTrue(self._database.matches("..", "aa.bb.cc"))
self.assertFalse(self._database.matches("..", "aa.bb."))
self.assertFalse(self._database.matches("..", "aa.bb"))
self.assertFalse(self._database.matches("..", "aa"))
self.assertTrue(self._database.matches("..#", "aa.bb.cc"))
self.assertFalse(self._database.matches("..#", "aa.bb."))
self.assertFalse(self._database.matches("..#", "aa.bb"))
self.assertFalse(self._database.matches("..#", "aa"))
self.assertTrue(self._database.matches(".#", "aa.bb.cc"))
self.assertTrue(self._database.matches(".#", "aa.bb."))
self.assertTrue(self._database.matches(".#", "aa.bb"))
self.assertFalse(self._database.matches(".#", "aa"))
def test_wrap_routes(self):
self.fail()
wrap = self._database.wrap_routes()
self.assertTrue(ROUTE_1 in wrap)
self.assertTrue(ROUTE_2 in wrap)
def test_find_route(self):
self.fail()
message = AMQMessageFactory.get_instance(111).create_request_message(
"POST", "clever3-book.localhost",
"/aa/bb?cc=d", {}, ""
)
route = self._database.find_route(message.domain, message.path)
self.assertIsNotNone(route)
self.assertEqual(ROUTE_3, route.as_string())
route = self._database.find_route("clever3.book.localhost", "/three-hot-dogs")
self.assertIsNotNone(route)
self.assertEqual(ROUTE_3, route.as_string())
route = self._database.find_route("clever-5.book.localhost", "/three-hot-dogs")
self.assertIsNone(route)
def test_add_route(self):
self.fail()
origSize = len(self._database.get_routes())
route = AMQRouteFactory.from_string(ROUTE_7)
self._database.add_route(route)
self.assertEqual(origSize + 1, len(self._database.get_routes()))
def test_remove_route(self):
self.fail()
origSize = len(self._database.get_routes())
route = AMQRouteFactory.from_string(ROUTE_7)
self._database.add_route(route)
self.assertEqual(origSize + 1, len(self._database.get_routes()))
self._database.remove_route(route)
self.assertEqual(origSize, len(self._database.get_routes()))
+95 -12
View File
@@ -1,39 +1,122 @@
from typing import Set
from unittest import TestCase
import pytest
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.model.model import AMQRoute
from amqp.router.router_base import RouterBase
from amqp.router.utils import sanitize_as_rabbitmq_name
ROUTE_1 = "amq::amq1:clever1.book.localhost.#:5:0"
ROUTE_2 = "amq::amq2:clever2.book.localhost.#:5:0"
ROUTE_3 = "amq::amq3:clever3.book.localhost.#:5:0"
ROUTE_6 = "amq::amq6:clever6.book.localhost.#:5:0"
ROUTE_7 = "amq::amq7:clever7.book.localhost.#:5:0"
ROUTE_8 = "amq::amq8:clever8.book.localhost.#:5:0"
ROUTE_9 = "amq::amq9:clever9.book.localhost.#:5:0"
class TestRouterBase(TestCase):
@pytest.fixture(autouse=True)
def setup_each_test(self):
self.base = RouterBase()
self.factory = AMQMessageFactory.get_instance(111)
if len(self.base.get_routes()) == 0:
r1 = AMQRouteFactory.from_string(ROUTE_1)
r2 = AMQRouteFactory.from_string(ROUTE_2)
r3 = AMQRouteFactory.from_string(ROUTE_3)
self.base.add_route(r1)
self.base.add_route(r2)
self.base.add_route(r3)
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#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"))
def test_get_routing_key(self):
self.fail()
message = self.factory.create_request_message("POST", "test.me.localhost", "/aa/bb?cc=d", {}, "")
rk = self.base.get_routing_key(message)
self.assertEqual("test.me.localhost.aa.bb", rk)
def test_wrap_routes(self):
self.fail()
wrap = self.base.wrap_routes()
self.assertTrue(ROUTE_1 in wrap)
self.assertTrue(ROUTE_2 in wrap)
def test_wrap_own_routes(self):
self.fail()
_route: AMQRoute = AMQRouteFactory.from_string(ROUTE_7)
self.base.add_own_route(_route)
o_route = self.base.wrap_own_routes()
self.assertEqual(ROUTE_7, o_route)
def test_unwrap_route_list(self):
self.fail()
_routes = self.base.get_routes()
wrapped = self.base.wrap_routes()
unwrapped = self.base.unwrap_route_list(wrapped)
self.assertEqual(len(_routes), len(unwrapped))
self.assertTrue(unwrapped.pop() in _routes)
self.assertTrue(unwrapped.pop() in _routes)
def test_get_routes(self):
self.fail()
_routes = self.base.get_routes()
self.assertEqual(3, len(_routes))
def test_find_route(self):
self.fail()
_route = self.base.find_route("clever3.book.localhost", "/three-hot-dogs")
self.assertIsNotNone(_route)
self.assertEqual(_route.as_string(), ROUTE_3)
_route = self.base.find_route("clever-5.book.localhost", "/three-hot-dogs")
self.assertIsNone(_route)
def test_find_route_by_message(self):
self.fail()
message = AMQMessageFactory.get_instance(111).create_request_message(
"POST", "clever3-book.localhost",
"/aa/bb?cc=d", {}, ""
)
_route: AMQRoute = self.base.find_route_by_message(message)
self.assertIsNotNone(_route)
self.assertEqual(_route.as_string(), ROUTE_3)
def test_add_routes(self):
self.fail()
init_routes: Set[AMQRoute] = self.base.get_routes().copy()
self.base.add_routes(ROUTE_6 + ",amq-dlq:DLX:DeadLetterQueue:#:0:0,:::::,:::")
modified = self.base.get_routes()
self.assertEqual(len(init_routes)+1, len(modified))
wr = self.base.wrap_routes()
self.assertFalse("DLX" in wr)
def test_remove_routes(self):
self.fail()
init_routes: Set[AMQRoute] = self.base.get_routes().copy()
toRemove: AMQRoute = init_routes.pop()
self.base.remove_routes(toRemove.as_string())
modified = self.base.get_routes()
self.assertEqual(len(init_routes), len(modified))
def test_remove_route(self):
self.fail()
origSize = len(self.base.get_routes())
_route = AMQRouteFactory.from_string(ROUTE_7)
self.base.add_route(_route)
self.assertEqual(origSize + 1, len(self.base.get_routes()))
self.base.remove_route(_route)
self.assertEqual(origSize, len(self.base.get_routes()))
def test_add_route(self):
self.fail()
origSize = len(self.base.get_routes())
_route = AMQRouteFactory.from_string(ROUTE_7)
self.base.add_route(_route)
self.assertEqual(origSize + 1, len(self.base.get_routes()))
def test_add_own_route(self):
self.fail()
_route = AMQRouteFactory.from_string(ROUTE_8)
self.base.add_own_route(_route)
self.assertTrue(ROUTE_8 in self.base.wrap_own_routes())
+2 -1
View File
@@ -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"))
+13 -3
View File
@@ -1,9 +1,19 @@
import re
pattern = re.compile(r"[^a-zA-Z0-9.#/\-]")
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)
return sanitize_as_rabbitmq_domain_name(token)
def sanitize_as_rabbitmq_domain_name(token: str) -> str:
# Step 1: Replace non-alphanumeric characters (except periods) with a period
step1 = re.sub(r'[^A-Za-z0-9.#]', '.', token)
# Step 2: Replace multiple consecutive periods with a single period
step2 = re.sub(r'\.+', '.', step1)
# Step 3: Remove trailing '.' character, if applicable
step_last_char = len(step2) - 1
return step2[:-1] if step_last_char > 0 and step2[step_last_char] == '.' else step2
+74 -51
View File
@@ -1,84 +1,107 @@
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
import logging.config
import os
import asyncio
from asyncio import Future
from aio_pika.abc import AbstractRobustExchange
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.config.amq_configuration import AMQConfiguration
from amqp.model.model import AMQMessage
from amqp.rabbitmq.rabbit_mq_consumer import RabbitMQConsumer
from amqp.service.service_message_handler import AMQServiceMessageHandler
from amqp.service.tracing import initialize_trace
logging.basicConfig(level=logging.DEBUG)
logging.info(f"CWD = {os.getcwd()}")
logging_conf_path = os.path.join(os.getcwd(), 'amqp', 'service', 'logging.conf')
if os.path.exists(logging_conf_path):
logging.config.fileConfig(logging_conf_path)
else:
logging_conf_path = os.path.join(os.getcwd(), 'logging.conf')
if os.path.exists(logging_conf_path):
logging.config.fileConfig(logging_conf_path)
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, service_adapter):
logging.info("***********************************************")
logging.info("********** AMQ Service *******************")
logging.info("***********************************************")
loop = asyncio.get_event_loop()
self.setup_opentelemetry()
self.register_routes()
self.register_rpc_handler()
self.rabbit_mq_consumer.request_routes()
self.amq_configuration = amq_configuration
self.message_factory = AMQMessageFactory.get_instance(self.amq_configuration.amq_adapter.generator_id)
self.rabbit_mq_consumer = RabbitMQConsumer(self.amq_configuration)
self.tracer = initialize_trace(name=amq_configuration.amq_adapter.service_name)
self.amq_service_message_handler = None
self.service_message_factory = CleverMicroServiceMessageFactory(
self.amq_configuration.amq_adapter.service_name
)
# self.init(loop) will initialize RabbitMQ connection
loop.run_until_complete(self.init(loop, service_adapter))
# loop.run_until_complete(consume(loop))
# self.consumer_thread = asyncio.create_task(self.rabbit_mq_consumer.channel.start_consuming())
# self.rabbit_mq_consumer.request_routes()
logging.info("***********************************************")
logging.info("****** AMQ Service Init Complete *********")
logging.info("***********************************************")
loop.run_forever()
async def init(self, loop, service_adapter):
exchange: AbstractRobustExchange = await self.rabbit_mq_consumer.init(loop)
self.amq_service_message_handler = AMQServiceMessageHandler(
service_adapter, self.tracer, self.message_factory, exchange
)
await self.register_routes()
await 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()
async def reinitialize_if_disconnected(self):
if not self.rabbit_mq_consumer.channel or self.rabbit_mq_consumer.channel.is_closed:
self.rabbit_mq_consumer = RabbitMQConsumer(self.amq_configuration)
await self.rabbit_mq_consumer.init(loop=None)
await self.register_routes()
await self.rabbit_mq_consumer.request_routes()
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()
async def register_routes(self) -> AbstractRobustExchange | None:
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)
await self.rabbit_mq_consumer.register_inbound_routes(
route_mapping, self.amq_service_message_handler.inbound_message_callback
)
self.rabbit_mq_consumer.register_rpc_handler(
self.amq_service_message_handler.on_reply_consumer(self.rabbit_mq_consumer.channel)
return await self.rabbit_mq_consumer.register_rpc_handler(
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)
return None
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('../config/application.properties.local')
# amq_service: AMQService = AMQService(amq_configuration, CleverSwarmAmqpAdapter(amq_configuration))
# amq_service.rabbit_mq_consumer.channel.start_consuming()
# amq_service.consumer_thread.cancel()
# amq_service.rabbit_mq_consumer.connection.close()
+34
View File
@@ -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
+50
View File
@@ -0,0 +1,50 @@
import io
import json
import logging
import python_multipart
from fastapi import UploadFile
from python_multipart import multipart
from python_multipart.multipart import Field, File
from amqp.model.model import AMQMessage
class CleverMultiPartParser:
def __init__(self, message: AMQMessage):
self.form_data = {}
self.is_multipart = False
self.is_json = False
self.is_valid = True
# Convert each list of strings to a single string joined by newline, then to bytes
headers = {key: "\n".join(value).encode("utf-8") for key, value in message.headers.items()}
content_type: str | bytes | None = headers.get("Content-Type")
content_type, params = multipart.parse_options_header(content_type)
if content_type.decode('utf-8') in ["application/octet-stream",
"multipart/form-data",
"application/x-www-form-urlencoded",
"application/x-url-encoded"]:
try:
python_multipart.parse_form(headers, io.BytesIO(message.body()), self.on_field, self.on_file)
self.is_multipart = True
except Exception as e:
logging.error(f"Error parsing multipart/form-data: {e}")
try:
self.form_data = json.loads(message.body())
self.is_json = True
except Exception as jsonerror:
logging.error(f"Error parsing JSON: {jsonerror}")
self.is_valid = False
else:
self.form_data = message.body()
def on_field(self, field: Field):
print(field)
self.form_data[field.field_name.decode('utf-8')] = field.value
print(self.form_data)
def on_file(self, file: File):
print(file)
self.form_data[file.field_name.decode('utf-8')] = \
UploadFile(file.file_object, size=file.size, filename=file.file_name)
print(self.form_data)
+148 -105
View File
@@ -1,66 +1,118 @@
import asyncio
import logging
import time
import concurrent.futures
import os
from asyncio import Future
from typing import Dict
from aio_pika import IncomingMessage, Message
from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange
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 amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.model.model import AMQResponse
from amqp.adapter.amq_response_factory import AMQResponseFactory
from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter
from amqp.model.model import AMQResponse, AMQMessage
from amqp.router.serializer import map_as_string
parallel_workers = os.getenv("PARALLEL_WORKERS", default=10)
# Shared thread pool for parallel execution
executor = concurrent.futures.ThreadPoolExecutor(max_workers=parallel_workers)
def collect_trace_info():
trace_map = {}
TraceContextTextMapPropagator().inject(
carrier=trace_map,
context=trace.context_api.get_current()
)
return trace_map
def get_default_trace_state():
return TraceState(entries={})
def set_text(carrier, key, value):
carrier[key] = value
class AMQServiceMessageHandler:
def __init__(self, http_adapter, tracer):
self.http_adapter = http_adapter
def __init__(self, service_adapter: CleverThisServiceAdapter,
tracer: Tracer, message_factory: AMQMessageFactory,
reply_to_exchange: AbstractRobustExchange):
self.service_adapter: CleverThisServiceAdapter = service_adapter
self.tracer = tracer
self.outstanding = {}
self.message_factory: AMQMessageFactory = message_factory
self.reply_to_exchange: AbstractRobustExchange = reply_to_exchange
# Dictionary to store outstanding Futures
self.outstanding: Dict[str, asyncio.Future] = {}
def on_delivery_callback_http(self, channel, consumerTag, delivery):
async def inbound_message_callback(self, message: AbstractIncomingMessage):
"""
Receives the message as bytes from RabbitMQ queue and deserializes it to the AMQMessage
Ensures Jaeger trace-id propagation.
Invokes custom handler/mapper/adapter method that corresponds to this message
:param message:
:return:
"""
start = time.time()
deliveryTag = delivery.envelope.delivery_tag
debug = f" [DBG] ######[{deliveryTag}] (AMQ MESSAGE), env: {delivery.envelope}, props: {delivery.properties}"
delivery_tag = message.delivery_tag
debug = f" [DBG] ######[{delivery_tag}] (AMQ MESSAGE), env: {message.headers}, props: {message.properties}"
logging.info(debug)
amq_message = AMQMessageFactory.from_bytes(delivery.body)
amq_message: AMQMessage = AMQMessageFactory.from_bytes(message.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, message.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:
future = executor.submit(self.service_adapter.sync_on_message, amq_message)
# response: AMQResponse = asyncio.run(self.http_adapter.on_message(amq_message))
# Wait for the result and send it back to the reply queue
try:
response: AMQResponse = future.result() # Block until the result is available
await self.send_reply(message.reply_to, delivery_tag, response)
except Exception as e:
print(f"[E] Error processing message: {e}")
await message.nack(requeue=False)
return
else:
logging.error(" [E] Unknown MAGIC value: [%s], don't know how to interpret the message",
amq_message.magic)
await message.nack(requeue=False)
return
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)
await message.ack(multiple=False)
def on_http_response(self, delivery, deliveryTag, channel, http_response):
"""
In loose coupling mode, receives the HTTP response to the REST request the HTTPAdapter made to the
associated CleverXXX service. Convert the response to AMQResponse message and use
RabbitMQ RPC reply-to mechanism to deliver this response back to API Gateway side AMQProducer.
:param delivery:
:param deliveryTag:
:param channel:
:param http_response:
:return:
"""
response = AMQResponse(
delivery.properties.correlation_id,
http_response.status_code,
@@ -68,76 +120,67 @@ 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)
self.send_reply(delivery.properties.reply_to, deliveryTag, 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
async def send_reply(self, reply_to: str, delivery_tag: int | None, response: AMQResponse):
"""
The Service side code. When the CleverXXX service has output ready, in AMQResponse object,
call this method to return the reply back to the caller
:param reply_to:
:param delivery_tag:
:param response:
:return:
"""
if reply_to:
logging.info(" [*] ######[%s] AMQ Message Reply, To=%s, msgId=%s",
delivery_tag, reply_to, response.id)
pika_message: Message = Message(
body=AMQResponseFactory.serialize(response),
correlation_id=response.id,
content_type=response.content_type[0]
if isinstance(response.content_type, list) else response.content_type
)
await self.reply_to_exchange.publish(
message=pika_message,
routing_key=reply_to
)
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)
)
async def reply_received_callback(self, message: IncomingMessage):
"""
The producer/client (API Gateway) facing response_message recipient - must match the received response with
the original request and produce the HTTP reply.
Note: this code runs on the API Gateway side of RabbitMQ
:param message:
:return:
"""
reply_id = message.correlation_id
content_type = message.content_type
delivery_tag = message.delivery_tag
debug = f" [DBG] ######[{delivery_tag}] (RPC REPLY) env: {message.headers}, props: {message.properties}, BODY: {message.body.decode()}"
logging.info(debug)
# TODO - figure out implementation of this part, as this is sending reply back either to a service,
# or to the user
logging.info(f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}")
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(message.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
await message.ack(multiple=False)
+47
View File
@@ -0,0 +1,47 @@
import json
import struct
import io
import pika
class StreamParser:
def __init__(self):
self.buffer = io.BytesIO() # Buffer to accumulate chunks
self.metadata_length = None # Length of the metadata
self.metadata = None # Parsed JSON metadata
self.file_writer = None # File writer for the remaining data
def process_chunk(self, chunk):
"""
Process a chunk of data and handle it according to the rules.
"""
self.buffer.write(chunk)
# If metadata length is not yet known, try to read it
if self.metadata_length is None and self.buffer.tell() >= 4:
self.buffer.seek(0)
self.metadata_length = struct.unpack('>I', self.buffer.read(4))[0]
# If metadata length is known but metadata is not yet parsed, try to parse it
if self.metadata_length is not None and self.metadata is None:
if self.buffer.tell() >= 4 + self.metadata_length:
self.buffer.seek(4)
metadata_bytes = self.buffer.read(self.metadata_length)
self.metadata = json.loads(metadata_bytes.decode('utf-8'))
print("Metadata:", self.metadata)
# Prepare to write the remaining data to a file
self.file_writer = open(self.metadata.get("filename", "output_file"), "wb")
# If metadata is parsed, write the remaining data to the file
if self.metadata is not None:
remaining_data = self.buffer.read()
if remaining_data:
self.file_writer.write(remaining_data)
def close(self):
"""
Close the file writer and clean up.
"""
if self.file_writer:
self.file_writer.close()
self.buffer.close()
+25 -14
View File
@@ -1,7 +1,6 @@
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 import TracerProvider, Tracer
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
@@ -11,7 +10,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 +22,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 +29,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 +72,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 +95,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)
Executable
+244
View File
@@ -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
@@ -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'
@@ -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": ""
}
+9
View File
@@ -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}
@@ -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}"
}
}
+21
View File
@@ -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"
+32
View File
@@ -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"
+1
View File
@@ -0,0 +1 @@
[rabbitmq_management,rabbitmq_prometheus].
+20
View File
@@ -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"
+20
View File
@@ -0,0 +1,20 @@
[build-system]
requires = ["setuptools>=61.0", "wheel"] # Specifies build tools to use
build-backend = "setuptools.build_meta"
[project]
name = "amq_adapter"
version = "0.2.6"
description = "The Python implementation of the AMQ Adaptor for CleverMicro"
readme = "README.md"
license = { text = "Apache License 2.0" }
dependencies = [
"aiohttp~=3.11.11",
"pika~=1.3.2",
"fastapi==0.115.6",
"opentelemetry-api",
"opentelemetry-sdk",
"opentelemetry-exporter-otlp",
"opentelemetry-exporter-prometheus",
"opentelemetry-instrumentation-pika"
]
+3 -3
View File
@@ -1,8 +1,8 @@
aiohttp~=3.11.11
pika~=1.3.2
flask
fastapi==0.115.6
aio_pika~=9.5.5
opentelemetry-api
opentelemetry-sdk
opentelemetry-exporter-otlp
opentelemetry-exporter-prometheus
opentelemetry-instrumentation-flask
opentelemetry-instrumentation-pika
+30
View File
@@ -0,0 +1,30 @@
from setuptools import setup, find_packages
setup(
name="amqp",
version="0.2.6",
author="Stanislav Hejny",
author_email="stanislav.hejny@cleverthis.com",
description="Adapter for RabbitMQ to integrate a CleverThis python-code Service with CleverMicro platform",
long_description=open("README.md").read(),
long_description_content_type="text/markdown",
url="https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python",
packages=find_packages(),
install_requires=[
# dependencies here
"aiohttp~=3.11.11",
"fastapi==0.115.6",
"pika~=1.3.2",
"opentelemetry-api",
"opentelemetry-sdk",
"opentelemetry-exporter-otlp",
"opentelemetry-exporter-prometheus",
"opentelemetry-instrumentation-pika"
],
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
],
python_requires=">=3.11",
)