Stanislav Hejny
263 lines
11 KiB
Python
263 lines
11 KiB
Python
import asyncio
|
|
import json
|
|
import time
|
|
from asyncio import AbstractEventLoop
|
|
from threading import Thread
|
|
|
|
from aio_pika import Message
|
|
from aio_pika.abc import AbstractRobustChannel
|
|
|
|
from amqp.adapter.logging_utils import logging_info, logging_warning
|
|
from amqp.config.amq_configuration import AMQConfiguration
|
|
from amqp.model.model import ScalingRequestAlert
|
|
from amqp.router.utils import await_future, await_result
|
|
|
|
|
|
class ScaleRequestV1:
|
|
|
|
def __init__(
|
|
self,
|
|
serviceId: str,
|
|
taskId: str,
|
|
maxAvailability: int,
|
|
currentAvailability: int,
|
|
requestType: ScalingRequestAlert,
|
|
):
|
|
self.version = 1 # Version of the request, currently 1
|
|
self.serviceId = serviceId
|
|
self.taskId = taskId
|
|
self.maxAvailability = maxAvailability
|
|
self.currentAvailability = currentAvailability
|
|
self.requestType = requestType
|
|
self.thread = None
|
|
|
|
def to_json(self) -> str:
|
|
"""Converts the object to a JSON string matching the Java structure"""
|
|
return json.dumps(
|
|
{
|
|
"version": self.version,
|
|
"serviceId": self.serviceId,
|
|
"taskId": self.taskId,
|
|
"maxAvailability": self.maxAvailability,
|
|
"currentAvailability": self.currentAvailability,
|
|
"requestType": self.requestType.value, # Using .value for Enum serialization
|
|
},
|
|
indent=2,
|
|
)
|
|
|
|
|
|
class BackpressureHandler:
|
|
# Track the number of messages currently being processed
|
|
current_parallel_executions = 0
|
|
# helps detect IDLE condition
|
|
last_data_message_time = 0
|
|
# helps prevent flooding the system with backpressure events
|
|
last_backpressure_event_time = 0
|
|
last_backpressure_event = ScalingRequestAlert.UPDATE
|
|
|
|
def __init__(
|
|
self,
|
|
channel: AbstractRobustChannel,
|
|
loop: AbstractEventLoop,
|
|
config: AMQConfiguration,
|
|
):
|
|
self.lock = None
|
|
self.channel = channel
|
|
self.loop = loop
|
|
self.config = config
|
|
self.exchange = None
|
|
self.swarm_service_id = self.config.amq_adapter.swarm_service_id
|
|
self.swarm_task_id = self.config.amq_adapter.swarm_task_id
|
|
self.parallel_workers = self.config.backpressure.threshold_threads
|
|
|
|
def increase_parallel_executions(self):
|
|
"""Increase the number of parallel executions"""
|
|
logging_info(
|
|
"Backpressure: Increase parallel executions, current=%s",
|
|
self.current_parallel_executions,
|
|
)
|
|
self.current_parallel_executions += 1
|
|
|
|
def decrease_parallel_executions(self):
|
|
"""Decrease the number of parallel executions"""
|
|
logging_info(
|
|
"Backpressure: Decrease parallel executions, current=%s",
|
|
self.current_parallel_executions,
|
|
)
|
|
if self.current_parallel_executions > 0:
|
|
self.current_parallel_executions -= 1
|
|
|
|
def update_parallel_executions(self, count: int):
|
|
"""Update the number of parallel executions"""
|
|
self.current_parallel_executions = count
|
|
|
|
def update_last_data_message_time(self):
|
|
logging_info("Backpressure: Update last data message time")
|
|
"""Update the last data message time"""
|
|
self.last_data_message_time = time.time()
|
|
|
|
def start_backpressure_monitor(self) -> Thread:
|
|
# Start the Backpressure monitor loop
|
|
self.thread = Thread(target=self.backpressure_monitor_loop)
|
|
self.thread.start()
|
|
return self.thread
|
|
|
|
async def check_overload_condition(self):
|
|
"""Check if the current parallel executions exceed the limit"""
|
|
self.update_last_data_message_time()
|
|
logging_info(
|
|
"Backpressure: Check overload condition, current=%s, max=%s",
|
|
self.current_parallel_executions,
|
|
self.parallel_workers,
|
|
)
|
|
if self.current_parallel_executions >= self.parallel_workers - 1:
|
|
# Check if the last backpressure event was not an overload
|
|
if self.last_backpressure_event != ScalingRequestAlert.OVERLOAD:
|
|
# Trigger the overload event
|
|
await self.handle_backpressure_overload_event()
|
|
self.last_backpressure_event_time = time.time()
|
|
self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
|
|
|
|
def backpressure_monitor_loop(self):
|
|
_time_window_sec = self.config.backpressure.time_window
|
|
_idle_duration_millis = 1000 * self.config.backpressure.idle_duration
|
|
_loop = asyncio.new_event_loop()
|
|
|
|
async def _backpressure_monitor():
|
|
"""Monitor the backpressure conditions and trigger events accordingly"""
|
|
while True:
|
|
logging_info(
|
|
"Backpressure: Monitor loop, current=%s",
|
|
self.current_parallel_executions,
|
|
)
|
|
logging_info(
|
|
"Backpressure: Last data message time=%s, eventTime=%s",
|
|
self.last_data_message_time,
|
|
self.last_backpressure_event_time,
|
|
)
|
|
logging_info(
|
|
"Backpressure: Last backpressure event=%s",
|
|
self.last_backpressure_event,
|
|
)
|
|
# Check if the current time exceeds the last backpressure event time
|
|
if (
|
|
self.current_parallel_executions >= self.parallel_workers
|
|
and time.time() - self.last_backpressure_event_time
|
|
> _time_window_sec
|
|
and self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
|
|
):
|
|
# Trigger the overload event
|
|
await self.handle_backpressure_overload_event()
|
|
self.last_backpressure_event_time = time.time()
|
|
self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
|
|
elif (
|
|
self.current_parallel_executions == 0
|
|
and time.time() - self.last_data_message_time
|
|
>= _idle_duration_millis
|
|
and self.last_backpressure_event != ScalingRequestAlert.IDLE
|
|
):
|
|
# Trigger the idle event
|
|
await self._handle_backpressure_idle_event()
|
|
self.last_backpressure_event = ScalingRequestAlert.IDLE
|
|
self.last_backpressure_event_time = time.time()
|
|
else:
|
|
# Trigger update event
|
|
if (
|
|
self.current_parallel_executions > 0
|
|
and time.time() - self.last_backpressure_event_time
|
|
> _time_window_sec
|
|
):
|
|
# Trigger the update event
|
|
_scaling_request: ScaleRequestV1 = ScaleRequestV1(
|
|
self.swarm_service_id,
|
|
self.swarm_task_id,
|
|
self.parallel_workers,
|
|
self.parallel_workers - self.current_parallel_executions,
|
|
ScalingRequestAlert.UPDATE,
|
|
)
|
|
# Address the message to any adapter capable of supporting BACKPRESSURE request
|
|
await self.publish_backpressure_request(_scaling_request)
|
|
self.last_backpressure_event_time = time.time()
|
|
self.last_backpressure_event = ScalingRequestAlert.UPDATE
|
|
await asyncio.sleep(_time_window_sec)
|
|
|
|
_loop.run_until_complete(_backpressure_monitor())
|
|
|
|
async def handle_backpressure_overload_event(self):
|
|
logging_warning("Backpressure: Capacity close to depleted!")
|
|
# Send an Overload event to the Management service
|
|
scaling_request: ScaleRequestV1 = ScaleRequestV1(
|
|
self.swarm_service_id,
|
|
self.swarm_task_id,
|
|
self.parallel_workers,
|
|
self.parallel_workers - self.current_parallel_executions,
|
|
ScalingRequestAlert.OVERLOAD,
|
|
)
|
|
# Address the message to any management service instance capable of supporting BACKPRESSURE request
|
|
await self.publish_backpressure_request(scaling_request)
|
|
# update / reset time-window so that the OVERLOAD is not sent too often
|
|
self.last_data_message_time = time.time()
|
|
|
|
async def _handle_backpressure_idle_event(self):
|
|
logging_warning("Backpressure: Service is idle.")
|
|
# Send an Idle event to the Management service
|
|
scaling_request: ScaleRequestV1 = ScaleRequestV1(
|
|
self.swarm_service_id,
|
|
self.swarm_task_id,
|
|
self.parallel_workers,
|
|
self.parallel_workers - self.current_parallel_executions,
|
|
ScalingRequestAlert.IDLE,
|
|
)
|
|
# Address the message to any adapter capable of supporting BACKPRESSURE request
|
|
await self.publish_backpressure_request(scaling_request)
|
|
# update / reset time-window so that the IDLE is not sent too often
|
|
self.last_data_message_time = time.time()
|
|
|
|
async def publish_backpressure_request(self, scaling_request: ScaleRequestV1):
|
|
# Publish the backpressure request to the management service
|
|
logging_info(
|
|
f"Publishing backpressure for {scaling_request.serviceId} with request type {scaling_request.requestType}"
|
|
)
|
|
if not self.exchange:
|
|
|
|
async def _wrap_rabbit_mq_api_init(channel):
|
|
_exchange = await channel.get_exchange(
|
|
name="cleverthis.clevermicro.management"
|
|
)
|
|
return _exchange
|
|
|
|
self.exchange = await await_result(
|
|
asyncio.run_coroutine_threadsafe(
|
|
_wrap_rabbit_mq_api_init(self.channel), self.loop
|
|
)
|
|
)
|
|
|
|
if self.exchange:
|
|
|
|
async def _wrap_rabbit_mq_api():
|
|
if not self.channel.is_closed:
|
|
binary_content: bytes = scaling_request.to_json().encode("utf-8")
|
|
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 self.exchange.publish(
|
|
message=pika_message, routing_key="backpressure-scaling-v1"
|
|
)
|
|
logging_info(
|
|
"Service Message Published to %s, msg: %s",
|
|
self.exchange.name,
|
|
str(binary_content),
|
|
)
|
|
return True
|
|
return False
|
|
|
|
await await_future(
|
|
asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop)
|
|
)
|