Service Module

class amqp.service.amq_service.AMQService(amq_configuration, service_adapter=None)[source]

Bases: object

Parameters:: amq_configuration (AMQConfiguration)

CLEVER_AMQP_CLIENT_VERSION = None

MAX_WAIT = 300

__init__(amq_configuration, service_adapter=None)[source]

Parameters:: amq_configuration (AMQConfiguration)

run()[source]: Start the AMQ service, initializing RabbitMQ connection and message handlers.

backpressure(current_availability, maximum=-1)[source]

Send Backpressure event to the Management Service. The Backpressure logic determines the type of the event based on the current_availability and maximum values. - OVERLOAD event is triggered immediately when the current_availability value is below threshold typically set to 10% of the maximum value, or 0 if no maximum is provided. - IDLE event is triggered when the current_availability value exeeds 80% of the maximum value or is greater than 0 if no maximum is provided. - UPDATE event is sent if neither OVERLOAD nor IDLE conditions are met, indicating that the backpressure is within acceptable limits. :param current_availability: Currently available capacity. :param maximum: Maximum value of the backpressure metric. If undefined or set to lt 0 (e.g. -1), the maximum is than determined from max value of current_availability seen over the time.

Parameters:

current_availability (int)
maximum (int)

rpc(service_name, fq_method_name, **kwargs)[source]

Send a Remote Procedure Call (RPC) message to the AMQ service. :param service_name: The service name to receive the RPC message. :param fq_method_name: The fully qualified method name to execute the RPC message. :param kwargs: Additional keyword arguments to be passed to the RPC method. :return: A Future that resolves when the RPC response is received.

Parameters:

service_name (str)
fq_method_name (str)

Return type:

Future

command(service_name, fq_method_name, **kwargs)[source]

Send a unidirectional Command message to the AMQ service. :param service_name: The service name to receive the RPC message. :param fq_method_name: The fully qualified method name to execute the RPC message. :param kwargs: Additional keyword arguments to be passed to the RPC method. :return: A Future that resolves when the RPC response is received.

Parameters:

service_name (str)
fq_method_name (str)

Return type:

bool

async data_message_generator(route)[source]

An async iterator that blocks until messages arrive, processes them, and continues in a loop.

Parameters:

queue_name – The name of the queue to consume messages from
route (AMQRoute)

Yield:

The processed DataResponse for each message

Return type:

AsyncIterator[DataMessage]

get_unique_instance_id()[source]

Provides cluster-wide unique instance ID for this AMQ service instance. This ID is used to identify the service instance in the cluster and is unique across all instances. IDs are managed by shared Consul KV store, ensuring that even if multiple instances of the service are running, they will each have unique ID.

Return type:: int

async init(loop, service_adapter)[source]

cleanup()[source]

async reinitialize_if_disconnected()[source]

async register_routes()[source]

Return type:: AbstractRobustExchange | None

set_outstanding_future(message, future)[source]

Parameters:

message (DataMessage)
future (Future)

remove_outstanding_future(message_id)[source]

Parameters:: message_id (str)

amqp.service.amq_message_handler.collect_trace_info()[source]

amqp.service.amq_message_handler.get_default_trace_state()[source]

amqp.service.amq_message_handler.set_text(carrier, key, value)[source]

async amqp.service.amq_message_handler.invoke_command(package_name, class_name, function_name, msg_id='', kwargs=None)[source]

Parameters:

package_name (str)
class_name (str)
function_name (str)
msg_id (str)

Return type:

Any

class amqp.service.amq_message_handler.DataMessageHandler(service_adapter, tracer, message_factory, service_message_factory, reply_to_exchange, rabbit_mq_client, amq_configuration, loop, backpressure_handler, file_handler=<amqp.adapter.file_handler.StreamingFileHandler object>)[source]

Bases: object

Parameters:

service_adapter (CleverThisServiceAdapter)
tracer (Tracer)
message_factory (DataMessageFactory)
service_message_factory (ServiceMessageFactory)
reply_to_exchange (AbstractRobustExchange)
rabbit_mq_client (RabbitMQClient)
amq_configuration (AMQConfiguration)
backpressure_handler (BackpressureHandler)
file_handler (FileHandler)

__init__(service_adapter, tracer, message_factory, service_message_factory, reply_to_exchange, rabbit_mq_client, amq_configuration, loop, backpressure_handler, file_handler=<amqp.adapter.file_handler.StreamingFileHandler object>)[source]

Parameters:

service_adapter (CleverThisServiceAdapter)
tracer (Tracer)
message_factory (DataMessageFactory)
service_message_factory (ServiceMessageFactory)
reply_to_exchange (AbstractRobustExchange)
rabbit_mq_client (RabbitMQClient)
amq_configuration (AMQConfiguration)
backpressure_handler (BackpressureHandler)
file_handler (FileHandler)

async inbound_data_message_callback_as_thread(message)[source]

The callback method that is called when a DataMessage is received from RabbitMQ, starts new Thread.

Parameters:: message (AbstractIncomingMessage)

async inbound_data_message_callback_no_thread(message)[source]

Receives the DataMessage as bytes from RabbitMQ queue and deserializes it to the DataMessage object Ensures Jaeger trace-id propagation. Invokes custom handler/mapper/adapter method that corresponds to this message :param message: :return:

Parameters:: message (AbstractIncomingMessage)

async run_http_request_magic(message, amq_message, loop)[source]

Synchronous version of on_message method. This is a wrapper around the async on_message method. It also counts parallel executions and handles backpressure. :param message: RabbitMQ raw message :param amq_message: DataMessage :param loop: Event loop :return: DataResponse

Parameters:

message (AbstractIncomingMessage)
amq_message (DataMessage)
loop (AbstractEventLoop)

Return type:

None

async run_rpc_request(message, amq_message, loop)[source]

Synchronous version of on_message method. This is a wrapper around the async on_message method. It also counts parallel executions and handles backpressure. :param message: RabbitMQ raw message :param amq_message: DataMessage :param loop: Event loop :return: DataResponse

Parameters:

message (AbstractIncomingMessage)
amq_message (DataMessage)
loop (AbstractEventLoop)

Return type:

None

async on_rpc(a_message)[source]

Dynamically invokes a function or a class method from a specified package.

Parameters:: a_message (AMQMessage) – The message containing the package, class, and function details.
Return type:: DataResponse

Args encoded in the AMQMessage:

package_name (str): The name of the Python package to import. class_name (str): The name of the class within the package.

Use ‘__global__’ if the function_name refers to a global function in the package.

function_name (str): The name of the function or method to invoke. kwargs (Dict[str, Any]): A dictionary of keyword arguments to pass to the function/method (in the body).

Returns:

The return value of the invoked function or method.

Return type:

Any

Raises:

ModuleNotFoundError – If the specified package cannot be imported.
AttributeError – If the specified class or function/method does not exist.
Exception – For any other errors during invocation.

Parameters:

a_message (AMQMessage)

async reconstitute_data_message(message)[source]

DataMessage may be serialized as string in separate queue (addressing == 1) or be a MultiPart message with files streamed over individual queues, one per file. This function uses ‘addressing’ mode from message headers to determine how to deserialize (reconstitute) the original message.

Parameters:: message (AbstractIncomingMessage)
Return type:: DataMessage | None

ensure_trace_info(amq_message)[source]

Parameters:: amq_message (DataMessage)

record_duration(start, delivery_tag)[source]

async send_reply(reply_to, response)[source]

The Service side code. When the CleverXXX service has output ready, in DataResponse object, call this method to return the reply back to the caller :param reply_to:connection.channel :param response: :return:

Parameters:

reply_to (str)
response (DataResponse)

async reply_received_callback(message)[source]

The handler for reply for DataMessage that we sent out earlier - the code matches the received response with the original request and passes the response to the caller. :param message: :return:

Parameters:: message (AbstractIncomingMessage)

async send_rpc(route, message)[source]

Sends an RPC message to the specified route and returns a Future that will be completed when the response is received.

Parameters:

route – AMQRoute object containing exchange and routing information
message (DataMessage) – DataMessage to be sent

Returns:

Future that will be completed with the response payload

Return type:

Future

async send_command(route, message)[source]

Sends an RPC message to the specified route and returns a Future that will be completed when the response is received.

Parameters:

route – AMQRoute object containing exchange and routing information
message (DataMessage) – DataMessage to be sent

Returns:

Future that will be completed with the response payload

Return type:

bool

OpenTelemetry configuration for tracing and metrics.

This module provides functionality to initialize and configure OpenTelemetry for both tracing (Jaeger via OTLP) and metrics (Prometheus) in the AMQ adapter.

class amqp.service.tracing.TracingConfig(service_name, service_version, otlp_endpoint='http://localhost:4317', insecure=True, debug=False, additional_attributes=None)[source]

Bases: object

Configuration for OpenTelemetry tracing.

Parameters:

service_name (str)
service_version (str)
otlp_endpoint (str)
insecure (bool)
debug (bool)
additional_attributes (Dict[str, Any] | None)

service_name: str

service_version: str

otlp_endpoint: str = 'http://localhost:4317'

insecure: bool = True

debug: bool = False

additional_attributes: Dict[str, Any] | None = None

__init__(service_name, service_version, otlp_endpoint='http://localhost:4317', insecure=True, debug=False, additional_attributes=None)

Parameters:

service_name (str)
service_version (str)
otlp_endpoint (str)
insecure (bool)
debug (bool)
additional_attributes (Dict[str, Any] | None)

Return type:

None

class amqp.service.tracing.MetricsConfig(prometheus_port=9464, prometheus_host='localhost', export_interval_millis=30000, additional_attributes=None)[source]

Bases: object

Configuration for OpenTelemetry metrics.

Parameters:

prometheus_port (int)
prometheus_host (str)
export_interval_millis (int)
additional_attributes (Dict[str, Any] | None)

prometheus_port: int = 9464

prometheus_host: str = 'localhost'

export_interval_millis: int = 30000

additional_attributes: Dict[str, Any] | None = None

__init__(prometheus_port=9464, prometheus_host='localhost', export_interval_millis=30000, additional_attributes=None)

Parameters:

prometheus_port (int)
prometheus_host (str)
export_interval_millis (int)
additional_attributes (Dict[str, Any] | None)

Return type:

None

amqp.service.tracing.create_resource(config)[source]

Create an OpenTelemetry resource with service information.

Parameters:: config (TracingConfig)
Return type:: Resource

amqp.service.tracing.setup_otlp_exporter(config)[source]

Configure and create an OTLP exporter for Jaeger.

Parameters:: config (TracingConfig)
Return type:: OTLPSpanExporter

amqp.service.tracing.setup_trace_provider(config, resource)[source]

Set up the TracerProvider with OTLP exporter.

Parameters:

config (TracingConfig)
resource (Resource)

Return type:

TracerProvider

amqp.service.tracing.setup_metrics_provider(config, resource)[source]

Set up the MeterProvider with Prometheus export.

Parameters:

config (MetricsConfig)
resource (Resource)

Return type:

MeterProvider

amqp.service.tracing.initialize_telemetry(tracing_config=None, metrics_config=None)[source]

Initialize OpenTelemetry with tracing and metrics.

Parameters:

tracing_config (TracingConfig | None) – Configuration for tracing
metrics_config (MetricsConfig | None) – Configuration for metrics

Returns:

Configured OpenTelemetry tracer

Return type:

Tracer

Example

```python config = TracingConfig(

service_name=”amq-adapter”, service_version=”1.0.0”, otlp_endpoint=”http://jaeger:4317”, debug=True

)

metrics_config = MetricsConfig(: prometheus_port=9464, prometheus_host=”0.0.0.0”

)

tracer = initialize_telemetry(config, metrics_config) ```

amqp.service.tracing.create_span(tracer, name, attributes=None)[source]

Create a new span with the given name and attributes.

Parameters:

tracer (Tracer) – The OpenTelemetry tracer
name (str) – Name of the span
attributes (Dict[str, Any] | None) – Optional attributes to add to the span

Returns:

A context manager that creates and manages a span