From 45941de6b15f8ea32894d12fafe55869524dca27 Mon Sep 17 00:00:00 2001 From: Rui Hu Date: Tue, 15 Jul 2025 15:07:39 +0800 Subject: [PATCH] fix(General): fix availability report Remove the loop for monitoring the CPU usage, adopt changes in the management service for the reporting message format, update the timing for sending out report, similar to java adapter. ISSUES RELATED: clevermicro/amq-adapter-python#58 --- .forgejo/workflows/build-and-publish.yml | 73 ------ .forgejo/workflows/publish-docker.yaml | 2 +- amqp/adapter/backpressure_handler.py | 205 +++------------ amqp/model/model.py | 14 - amqp/service/amq_message_handler.py | 10 +- amqp/service/amq_service.py | 19 -- tests/adapter/test_backpressure_handler.py | 285 +++------------------ tests/model/test_model.py | 9 - tests/service/test_amq_service.py | 33 ++- 9 files changed, 94 insertions(+), 556 deletions(-) delete mode 100644 .forgejo/workflows/build-and-publish.yml diff --git a/.forgejo/workflows/build-and-publish.yml b/.forgejo/workflows/build-and-publish.yml deleted file mode 100644 index dfec56b..0000000 --- a/.forgejo/workflows/build-and-publish.yml +++ /dev/null @@ -1,73 +0,0 @@ -name: Build and Publish Docker Image - -#on: -# push: -# branches: [ main, master ] -# tags: [ 'v*' ] -# pull_request: -# branches: [ main, master ] -on: - push: - branches: - - master # publish release on master - - develop # publish snapshot on develop - - feat-58-backpressure-reactive-streams - workflow_dispatch: - # allow manual trigger - -env: - REGISTRY_URL: "git.cleverthis.com" - REPOSITORY: "clevermicro/amq-adapter-python-demo" - DOCKER_HOST: "tcp://dind:2375" - -jobs: - build-and-push: - runs-on: ubuntu-latest - services: - dind: - image: docker:dind - cmd: - - dockerd - - -H - - tcp://0.0.0.0:2375 - - --tls=false - steps: - - name: Checkout code - uses: actions/checkout@v4 - with: - ref: ${{ github.ref_name }} - - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v2 - - - name: Login to Docker Registry - uses: docker/login-action@v3 - with: - registry: ${{ env.REGISTRY_URL }} - username: ${{ secrets.REGISTRY_USER }} - password: ${{ secrets.REGISTRY_PASSWORD }} - if: github.event_name != 'pull_request' - - - name: Extract metadata for Docker - id: meta - uses: docker/metadata-action@v4 - with: - images: ${{ env.REGISTRY_URL }}/clevermicro/amq-adapter-python-demo - tags: | - type=ref,event=branch - type=ref,event=pr - type=semver,pattern={{version}} - type=semver,pattern={{major}}.{{minor}} - type=sha,format=short - - - name: Build and push Docker image - uses: docker/build-push-action@v6 - with: - context: . - file: ./Dockerfile - # push: ${{ github.event_name != 'pull_request' }} - push: true - tags: ${{ env.REGISTRY_URL }}/${{ env.REPOSITORY }}:latest - labels: ${{ steps.meta.outputs.labels }} - cache-from: type=gha - cache-to: type=gha,mode=max diff --git a/.forgejo/workflows/publish-docker.yaml b/.forgejo/workflows/publish-docker.yaml index 8461141..b6b4d2a 100644 --- a/.forgejo/workflows/publish-docker.yaml +++ b/.forgejo/workflows/publish-docker.yaml @@ -38,4 +38,4 @@ jobs: file: ./Dockerfile no-cache: true push: true - tags: ${{ env.REGISTRY_URL }}/${{ env.REPOSITORY }}:20250709-01 + tags: ${{ env.REGISTRY_URL }}/${{ env.REPOSITORY }}:20250715-01 diff --git a/amqp/adapter/backpressure_handler.py b/amqp/adapter/backpressure_handler.py index ea0cf72..9be0f81 100644 --- a/amqp/adapter/backpressure_handler.py +++ b/amqp/adapter/backpressure_handler.py @@ -4,13 +4,11 @@ import time from asyncio import AbstractEventLoop from threading import Thread -import psutil from aio_pika import Message from aio_pika.abc import AbstractRobustChannel -from amqp.adapter.logging_utils import logging_info, logging_warning +from amqp.adapter.logging_utils import logging_info from amqp.config.amq_configuration import AMQConfiguration -from amqp.model.model import ScalingRequestAlert from amqp.router.utils import await_future, await_result CPU_IDLE, CPU_ACTIVE, CPU_OVERLOAD = 0, 1, 2 @@ -24,16 +22,12 @@ class ScaleRequestV1: 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: @@ -43,9 +37,7 @@ class ScaleRequestV1: "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, ) @@ -95,11 +87,10 @@ class RunningAverage: 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 + # report when remain availability goes down + last_usage_report_time = 0 + # report when remain availability goes up + last_recover_report_time = 0 # _callback_list is used in unit tests to record the invoked callbacks _callback_list = None @@ -117,12 +108,7 @@ class BackpressureHandler: 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 - self.average_cpu_usage = None self.do_loop = -1 - self._resource_usage_state = RESOURCE_UNSET - self._resource_usage_changed = 0 - self._resource_average_value = 0 - self._last_resource_max_value = 100 # Default max value for CPU usage def _do_loop(self) -> bool: """ @@ -135,177 +121,66 @@ class BackpressureHandler: self.do_loop -= 1 return _val - def increase_parallel_executions(self): + async 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", + "Backpressure: Increased parallel executions, new value: %s", self.current_parallel_executions, ) + # increase parallel -> reduce in remaining availability -> usage + # report usage at least every 1s + # TODO: add config for the interval + if self.last_usage_report_time + 1 < time.time(): + await self.send_backpressure_report() + self.last_usage_report_time = time.time() + + async def decrease_parallel_executions(self): + """Decrease the number of 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() + logging_info( + "Backpressure: Decreased parallel executions, new value=%s", + self.current_parallel_executions, + ) + # decrease parallel -> add to remaining availability -> recover + # report recover at every 3s + # TODO: add config for the interval + if self.last_recover_report_time + 3 < time.time(): + await self.send_backpressure_report() + self.last_recover_report_time = time.time() def start_backpressure_monitor(self) -> Thread: # Start the Backpressure monitor loop - self.thread = Thread(target=self.backpressure_monitor_loop) + self.thread = Thread(target=self.regular_report_loop_wrapper) self.thread.daemon = True # This makes it a daemon thread 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 - - async def _backpressure_monitor(self): - """Monitor the backpressure conditions and trigger events accordingly""" - _time_window_millis = self.config.backpressure.time_window - _idle_duration_millis = self.config.backpressure.idle_duration - _overload_duration_millis = self.config.backpressure.cpu_overload_duration - _monitor_interval = 0.5 # Monitor every 500ms - CPU_IDLE, CPU_ACTIVE, CPU_OVERLOAD = 0, 1, 2 - IDLE_THRESHOLD, OVERLOAD_THRESHOLD = 15, 90 - if self.average_cpu_usage is None: - self.average_cpu_usage = RunningAverage( - int(max(_overload_duration_millis, _idle_duration_millis) // _monitor_interval) - ) - _cpu_usage_state = CPU_ACTIVE - _cpu_usage_changed = time.time() + async def _regular_report_loop(self): + """Regularly report the availability in a fixed interval""" while self._do_loop(): - _current_cpu_usage = psutil.cpu_percent(interval=None) - self.average_cpu_usage.add(_current_cpu_usage) - _average_cpu_usage = self.average_cpu_usage.get_average() - _now = time.time() - # logging_info( - # "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s", - # _current_cpu_usage, - # _average_cpu_usage, - # _cpu_usage_state, - # self.last_data_message_time, - # self.last_backpressure_event, - # self.last_backpressure_event_time, - # ) - if _average_cpu_usage < IDLE_THRESHOLD: - if _cpu_usage_state != CPU_IDLE: - _cpu_usage_changed = _now - _cpu_usage_state = CPU_IDLE - elif _average_cpu_usage > OVERLOAD_THRESHOLD: - if _cpu_usage_state != CPU_OVERLOAD: - _cpu_usage_changed = _now - _cpu_usage_state = CPU_OVERLOAD - else: - if _cpu_usage_state != CPU_ACTIVE: - _cpu_usage_changed = _now - _cpu_usage_state = CPU_ACTIVE + await self.send_backpressure_report() + # report recover at every 5s + # TODO: add config for the interval + await asyncio.sleep(5) - # Check if the current time exceeds the last backpressure event time - if ( - _cpu_usage_state == CPU_OVERLOAD - and _now - _cpu_usage_changed > _overload_duration_millis - and ( - _now - self.last_backpressure_event_time > _time_window_millis - or self.last_backpressure_event != ScalingRequestAlert.OVERLOAD - ) - ): - # Trigger the overload event - await self.handle_backpressure_overload_event() - self.last_backpressure_event_time = _now - self.last_backpressure_event = ScalingRequestAlert.OVERLOAD - elif ( - _cpu_usage_state == CPU_IDLE - and _now - _cpu_usage_changed > _idle_duration_millis - and _now - self.last_data_message_time > _idle_duration_millis - and ( - _now - self.last_backpressure_event_time > _time_window_millis - or 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 = _now - else: - # Trigger update event - if _now - self.last_backpressure_event_time > _time_window_millis: - # Trigger the update event - _scaling_request: ScaleRequestV1 = ScaleRequestV1( - self.swarm_service_id, - self.swarm_task_id, - 100, - _average_cpu_usage, - 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 = _now - self.last_backpressure_event = ScalingRequestAlert.UPDATE - await asyncio.sleep(_monitor_interval) - - def backpressure_monitor_loop(self): + def regular_report_loop_wrapper(self): _loop = asyncio.new_event_loop() - _loop.run_until_complete(self._backpressure_monitor()) + _loop.run_until_complete(self._regular_report_loop()) - async def handle_backpressure_overload_event(self): - logging_warning("Backpressure: Capacity close to depleted!") - # Send an Overload event to the Management service + async def send_backpressure_report(self): scaling_request: ScaleRequestV1 = ScaleRequestV1( - self.swarm_service_id, - self.swarm_task_id, - self.parallel_workers, - self.parallel_workers - self.current_parallel_executions, - ScalingRequestAlert.OVERLOAD, + serviceId=self.swarm_service_id, + taskId=self.swarm_task_id, + currentAvailability=self.parallel_workers - self.current_parallel_executions, ) - # 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}" + f"Publishing backpressure for {scaling_request.serviceId} with availability {scaling_request.currentAvailability}" ) if not self.exchange: diff --git a/amqp/model/model.py b/amqp/model/model.py index 08742b3..c67f308 100644 --- a/amqp/model/model.py +++ b/amqp/model/model.py @@ -291,21 +291,11 @@ class ServiceMessage: return self.message_type is not None -class ScalingRequestAlert(Enum): - OVERLOAD = "OVERLOAD" - IDLE = "IDLE" - UPDATE = "UPDATE" - SCALE_UP = "SCALE_UP" - SCALE_DOWN = "SCALE_DOWN" - - @dataclass(frozen=True) class ScalingRequest: service_id: str task_id: str - max_availability: int current_availability: int - request_type: ScalingRequestAlert version: int = 1 def to_json(self) -> str: @@ -315,9 +305,7 @@ class ScalingRequest: "version": self.version, "serviceId": self.service_id, "taskId": self.task_id, - "maxAvailability": self.max_availability, "currentAvailability": self.current_availability, - "requestType": self.request_type.value, # Using .value for Enum serialization }, indent=2, ) @@ -329,9 +317,7 @@ class ScalingRequest: return cls( service_id=data["serviceId"], task_id=data["taskId"], - max_availability=data["maxAvailability"], current_availability=data["currentAvailability"], - request_type=ScalingRequestAlert(data["requestType"]), # Convert string to Enum ) diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index b30429c..d1b98e2 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -123,9 +123,8 @@ class DataMessageHandler: :param loop: Event loop :return: DataResponse """ - # Increment the counter for parallel executions and check resource availability - self.backpressure_handler.increase_parallel_executions() - await self.backpressure_handler.check_overload_condition() + # Increment the counter for parallel executions + await self.backpressure_handler.increase_parallel_executions() logging_info( f"PARALLEL: Current Capacity [{self.backpressure_handler.current_parallel_executions}] of [{self.backpressure_handler.parallel_workers}]" ) @@ -154,9 +153,8 @@ class DataMessageHandler: logging_error(f"Deleting file {filename}: {e}") except Exception as e: logging_error(f"Request handler: {e}") - - self.backpressure_handler.decrease_parallel_executions() - await self.backpressure_handler.check_overload_condition() + # recover availability + await self.backpressure_handler.decrease_parallel_executions() async def reconstitute_data_message( self, message: AbstractIncomingMessage diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py index a6bf2d6..980c237 100644 --- a/amqp/service/amq_service.py +++ b/amqp/service/amq_service.py @@ -79,25 +79,6 @@ class AMQService: logging.info("***********************************************") self.loop.run_forever() - def backpressure(self, current: int, maximum: int = -1): - """ - Send Backpressure event to the Management Service. The Backpressure logic determines the type of the event - based on the current and maximum values. - - OVERLOAD event is triggered immediately when the current value exceeds the maximum threshold, which is - typically set to 80% of the maximum value. - - IDLE event is triggered when the current value drops below 20% of the maximum value and remains there for - a IDLE_WINDOW period of time (see BackpressureHandler). - - UPDATE event is sent if neither OVERLOAD nor IDLE conditions are met, indicating that the backpressure is - within acceptable limits. - :param current: Current value of the backpressure metric. - :param maximum: Maximum value of the backpressure metric. If undefined or set to lt 0, the maximum is set - according to the configuration value 'cm.backpressure.threshold'. - """ - if self.backpressure_handler is not None: - if maximum < 0: - maximum = self.amq_configuration.backpressure.threshold_threads - self.backpressure_handler.update_backpressure_value(current, maximum) - def rpc(self, service_name: str, fq_method_name: str, **kwargs) -> Future: """ Send a Remote Procedure Call (RPC) message to the AMQ service. diff --git a/tests/adapter/test_backpressure_handler.py b/tests/adapter/test_backpressure_handler.py index d2b75e0..39c840e 100644 --- a/tests/adapter/test_backpressure_handler.py +++ b/tests/adapter/test_backpressure_handler.py @@ -13,15 +13,11 @@ from aio_pika import Exchange from aio_pika.abc import AbstractRobustChannel from amqp.adapter.backpressure_handler import ( - RESOURCE_ACTIVE, - RESOURCE_OVERLOAD, - RESOURCE_UNSET, BackpressureHandler, RunningAverage, ScaleRequestV1, ) from amqp.config.amq_configuration import AMQConfiguration -from amqp.model.model import ScalingRequestAlert class TestScaleRequestV1: @@ -35,24 +31,18 @@ class TestScaleRequestV1: """ service_id = "test-service" task_id = "test-task" - max_availability = 10 current_availability = 5 - request_type = ScalingRequestAlert.UPDATE scale_request = ScaleRequestV1( service_id, task_id, - max_availability, current_availability, - request_type, ) assert scale_request.version == 1 assert scale_request.serviceId == service_id assert scale_request.taskId == task_id - assert scale_request.maxAvailability == max_availability assert scale_request.currentAvailability == current_availability - assert scale_request.requestType == request_type assert scale_request.thread is None def test_scale_request_v1_to_json(self): @@ -61,16 +51,12 @@ class TestScaleRequestV1: """ service_id = "test-service" task_id = "test-task" - max_availability = 10 current_availability = 5 - request_type = ScalingRequestAlert.UPDATE scale_request = ScaleRequestV1( service_id, task_id, - max_availability, current_availability, - request_type, ) json_output = scale_request.to_json() @@ -79,9 +65,7 @@ class TestScaleRequestV1: "version": 1, "serviceId": service_id, "taskId": task_id, - "maxAvailability": max_availability, "currentAvailability": current_availability, - "requestType": request_type.value, }, indent=2, ) @@ -93,16 +77,12 @@ class TestScaleRequestV1: """ service_id = "test-service" task_id = "test-task" - max_availability = 10 current_availability = 5 - request_type = ScalingRequestAlert.OVERLOAD # Change the alert type scale_request = ScaleRequestV1( service_id, task_id, - max_availability, current_availability, - request_type, ) json_output = scale_request.to_json() @@ -111,9 +91,7 @@ class TestScaleRequestV1: "version": 1, "serviceId": service_id, "taskId": task_id, - "maxAvailability": max_availability, "currentAvailability": current_availability, - "requestType": request_type.value, # Ensure the value is correctly serialized }, indent=2, ) @@ -288,47 +266,43 @@ class TestBackpressureHandler: assert handler.swarm_service_id == "clever-amqp-service" assert handler.swarm_task_id == "python-adapter" assert handler.parallel_workers == mock_config.backpressure.threshold_threads - assert handler.average_cpu_usage is None - def test_increase_parallel_executions(self, mock_channel, mock_loop, mock_config): + @pytest.mark.asyncio + async def test_increase_parallel_executions(self, mock_channel, mock_loop, mock_config): """ Test the increase_parallel_executions method. """ handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.send_backpressure_report = AsyncMock() initial_executions = handler.current_parallel_executions - handler.increase_parallel_executions() - assert handler.current_parallel_executions == initial_executions + 1 - def test_decrease_parallel_executions(self, mock_channel, mock_loop, mock_config): + with patch("time.time") as mock_time: + mock_time.return_value = 12345 + await handler.increase_parallel_executions() + assert handler.current_parallel_executions == initial_executions + 1 + assert handler.last_usage_report_time == 12345 + assert handler.send_backpressure_report.call_count == 1 + + @pytest.mark.asyncio + async def test_decrease_parallel_executions(self, mock_channel, mock_loop, mock_config): """ Test the decrease_parallel_executions method. """ handler = BackpressureHandler(mock_channel, mock_loop, mock_config) + handler.send_backpressure_report = AsyncMock() handler.current_parallel_executions = 5 - handler.decrease_parallel_executions() - assert handler.current_parallel_executions == 4 - handler.current_parallel_executions = 0 - handler.decrease_parallel_executions() - assert handler.current_parallel_executions == 0 # Should not go below 0 - - def test_update_parallel_executions(self, mock_channel, mock_loop, mock_config): - """ - Test the update_parallel_executions method. - """ - handler = BackpressureHandler(mock_channel, mock_loop, mock_config) - handler.update_parallel_executions(100) - assert handler.current_parallel_executions == 100 - - def test_update_last_data_message_time(self, mock_channel, mock_loop, mock_config): - """ - Test the update_last_data_message_time method. - """ - handler = BackpressureHandler(mock_channel, mock_loop, mock_config) with patch("time.time") as mock_time: mock_time.return_value = 12345 - handler.update_last_data_message_time() - assert handler.last_data_message_time == 12345 + await handler.decrease_parallel_executions() + assert handler.current_parallel_executions == 4 + assert handler.last_recover_report_time == 12345 + assert handler.send_backpressure_report.call_count == 1 + + handler.last_recover_report_time = 0 + handler.current_parallel_executions = 0 + await handler.decrease_parallel_executions() + assert handler.current_parallel_executions == 0 # Should not go below 0 def test_start_backpressure_monitor(self, mock_channel, mock_loop, mock_config): """ @@ -341,222 +315,29 @@ class TestBackpressureHandler: assert handler.thread == thread @pytest.mark.asyncio - async def test_check_overload_condition_no_overload(self, mock_channel, mock_loop, mock_config): + async def test_regular_report_loop(self, mock_channel, mock_loop, mock_config): """ - Test the check_overload_condition method when there is no overload. + Test the _regular_report_loop method. """ handler = BackpressureHandler(mock_channel, mock_loop, mock_config) - handler.current_parallel_executions = 1 - handler.parallel_workers = 5 - handler.last_backpressure_event = ScalingRequestAlert.UPDATE - with patch.object(handler, "handle_backpressure_overload_event") as mock_handle_overload: - await handler.check_overload_condition() - mock_handle_overload.assert_not_called() + handler.send_backpressure_report = AsyncMock() - @pytest.mark.asyncio - async def test_check_overload_condition_overload(self, mock_channel, mock_loop, mock_config): - """ - Test the check_overload_condition method when there is an overload. - """ - handler = BackpressureHandler(mock_channel, mock_loop, mock_config) - handler.current_parallel_executions = 9 # Assuming parallel_workers is 10 - handler.last_backpressure_event = ScalingRequestAlert.UPDATE - with patch.object(handler, "handle_backpressure_overload_event") as mock_handle_overload: - with patch("time.time") as mock_time: - mock_time.return_value = 12345 - await handler.check_overload_condition() - mock_handle_overload.assert_called_once() - assert handler.last_backpressure_event_time == 12345 - assert handler.last_backpressure_event == ScalingRequestAlert.OVERLOAD - - @pytest.mark.asyncio - async def test_check_overload_condition_already_overload( - self, mock_channel, mock_loop, mock_config - ): - """ - Test the check_overload_condition method when the last event was already an overload. - """ - handler = BackpressureHandler(mock_channel, mock_loop, mock_config) - handler.current_parallel_executions = 10 - handler.last_backpressure_event = ScalingRequestAlert.OVERLOAD - with patch.object(handler, "handle_backpressure_overload_event") as mock_handle_overload: - await handler.check_overload_condition() - mock_handle_overload.assert_not_called() - - # backpressure_monitor_loop is difficult to test directly without significant mocking - # and potentially long-running tests. The following is a basic outline of how - # you might approach testing it, focusing on verifying that the correct methods - # are called under specific conditions. This is NOT a complete, runnable test. - @pytest.mark.asyncio - async def test_backpressure_monitor_loop(self, mock_channel, mock_loop, mock_config): - """ - Test the backpressure_monitor_loop method (outline). - """ - handler = BackpressureHandler(mock_channel, mock_loop, mock_config) - handler.average_cpu_usage = RunningAverage( - 100, initial_value=99 - ) # Mock the RunningAverage in OVERLOAD state - handler.average_cpu_usage.is_filled = True # Together with 0 values ensures OVERLOAD state - # Setup mocks for dependent methods - handler.handle_backpressure_overload_event = AsyncMock() - handler._handle_backpressure_idle_event = AsyncMock() - handler.publish_backpressure_request = AsyncMock() - - # 1. Simulate an overload condition: - handler.last_backpressure_event_time = 0 # Force event trigger - handler.last_data_message_time = 0 + # test 1 loop handler.do_loop = 1 - handler.config.backpressure.cpu_overload_duration = ( - -1 - ) # Force overload condition even for recent change - await handler._backpressure_monitor() - handler.handle_backpressure_overload_event.assert_called_once() - - # 2. Simulate an idle condition - handler.average_cpu_usage = RunningAverage( - 100, initial_value=0 - ) # Mock the RunningAverage in IDLE state - handler.average_cpu_usage.is_filled = True # Together with 0 values ensures IDLE state - handler.last_backpressure_event_time = 0 - handler.last_data_message_time = 0 - handler.do_loop = 1 - handler.config.backpressure.idle_duration = ( - -1 - ) # Force idle condition even for recent change - await handler._backpressure_monitor() - handler._handle_backpressure_idle_event.assert_called_once() - - # 3. Simulate neither idle nor overload, but time_window passed - handler.handle_backpressure_overload_event.call_count = 0 - handler._handle_backpressure_idle_event.call_count = 0 - handler.average_cpu_usage = RunningAverage(100) # Mock the RunningAverage in pritine state - handler.last_backpressure_event_time = 0 - handler.last_data_message_time = 0 - handler.do_loop = 1 - await handler._backpressure_monitor() - assert ( - handler.publish_backpressure_request.call_count - + handler.handle_backpressure_overload_event.call_count - + handler._handle_backpressure_idle_event.call_count - ) > 0 + await handler._regular_report_loop() + handler.send_backpressure_report.assert_called_once() @pytest.mark.asyncio - async def test_handle_backpressure_overload_event(self, mock_channel, mock_loop, mock_config): + async def test_send_backpressure_report(self, mock_channel, mock_loop, mock_config): """ - Test the handle_backpressure_overload_event method. + Test the send_backpressure_report method. """ handler = BackpressureHandler(mock_channel, mock_loop, mock_config) handler.publish_backpressure_request = AsyncMock() with patch("time.time") as mock_time: mock_time.return_value = 12345 - await handler.handle_backpressure_overload_event() + await handler.send_backpressure_report() assert handler.publish_backpressure_request.call_count == 1 - assert handler.last_data_message_time == 12345 - - @pytest.mark.asyncio - async def test_handle_backpressure_idle_event(self, mock_channel, mock_loop, mock_config): - """ - Test the _handle_backpressure_idle_event method. - """ - handler = BackpressureHandler(mock_channel, mock_loop, mock_config) - handler.publish_backpressure_request = AsyncMock() - with patch("time.time") as mock_time: - mock_time.return_value = 12345 - await handler._handle_backpressure_idle_event() - assert handler.publish_backpressure_request.call_count == 1 - assert handler.last_data_message_time == 12345 - - @pytest.mark.asyncio - async def test_check_resource_availability_overload(self, mock_channel, mock_loop, mock_config): - """ - Test the check_resource_availability method when resource usage is in OVERLOAD state. - """ - handler = BackpressureHandler(mock_channel, mock_loop, mock_config) - handler.handle_backpressure_overload_event = AsyncMock() - handler._handle_backpressure_idle_event = AsyncMock() - handler.publish_backpressure_request = AsyncMock() - - # Set up initial state - handler.last_backpressure_event = ScalingRequestAlert.UPDATE - handler.last_backpressure_event_time = 0 - handler._resource_usage_state = RESOURCE_ACTIVE # Start with ACTIVE state - handler._resource_usage_changed = 0 - - with patch("time.time") as mock_time: - mock_time.return_value = 100 - # Call with resource usage above OVERLOAD threshold (90% of maximum) - await handler.check_resource_availability(95, 100) - - # Debug output - print(f"Resource usage state: {handler._resource_usage_state}") - print(f"Last backpressure event: {handler.last_backpressure_event}") - print(f"Last backpressure event time: {handler.last_backpressure_event_time}") - print(f"Time window: {mock_config.backpressure.time_window}") - print("Resource usage: 95, Maximum: 100") - print(f"Overload threshold calculation: {round(0.9 * 100)}") - - # Verify OVERLOAD event was triggered - handler.handle_backpressure_overload_event.assert_called_once() - assert handler.last_backpressure_event == ScalingRequestAlert.OVERLOAD - assert handler.last_backpressure_event_time == 100 - assert handler._resource_usage_state == RESOURCE_OVERLOAD - - @pytest.mark.asyncio - async def test_check_resource_availability_idle(self, mock_channel, mock_loop, mock_config): - """ - Test the check_resource_availability method when resource usage is in IDLE state. - """ - handler = BackpressureHandler(mock_channel, mock_loop, mock_config) - handler._handle_backpressure_idle_event = AsyncMock() - handler.handle_backpressure_overload_event = AsyncMock() - handler.publish_backpressure_request = AsyncMock() - - # Set up initial state - handler.last_backpressure_event = ScalingRequestAlert.UPDATE - handler.last_backpressure_event_time = 0 - handler._resource_usage_state = RESOURCE_UNSET # Start with UNSET state - handler._resource_usage_changed = 0 - - with patch("time.time") as mock_time: - # First call to set state to IDLE - mock_time.return_value = 50 - await handler.check_resource_availability(10, 100) - - # Second call after idle_duration has passed - mock_time.return_value = 50 + mock_config.backpressure.idle_duration + 1 - await handler.check_resource_availability(10, 100) - - # Verify IDLE event was triggered - handler._handle_backpressure_idle_event.assert_called_once() - assert handler.last_backpressure_event == ScalingRequestAlert.IDLE - assert ( - handler.last_backpressure_event_time == 50 + mock_config.backpressure.idle_duration + 1 - ) - - @pytest.mark.asyncio - async def test_check_resource_availability_update(self, mock_channel, mock_loop, mock_config): - """ - Test the check_resource_availability method when resource usage is in ACTIVE state. - """ - handler = BackpressureHandler(mock_channel, mock_loop, mock_config) - handler.publish_backpressure_request = AsyncMock() - handler.handle_backpressure_overload_event = AsyncMock() - handler._handle_backpressure_idle_event = AsyncMock() - - # Set up initial state - handler.last_backpressure_event = ScalingRequestAlert.IDLE - handler.last_backpressure_event_time = 0 - - with patch("time.time") as mock_time: - mock_time.return_value = 100 - # Call with resource usage between thresholds - await handler.check_resource_availability(50, 100) - - # Verify UPDATE event was triggered - assert handler.publish_backpressure_request.call_count == 1 - assert handler.last_backpressure_event == ScalingRequestAlert.UPDATE - assert handler.last_backpressure_event_time == 100 - assert handler._resource_usage_state == RESOURCE_ACTIVE @pytest.mark.asyncio async def test_publish_backpressure_request_exchange_exists( @@ -571,7 +352,7 @@ class TestBackpressureHandler: mock_async_run.return_value = asyncio.Future() mock_async_run.return_value.set_result(None) await handler.publish_backpressure_request( - ScaleRequestV1("test_service", "test_task", 100, 50, ScalingRequestAlert.UPDATE) + ScaleRequestV1("test_service", "test_task", 50) ) assert mock_async_run.call_count == 1 # handler.exchange.publish.assert_called_once() @@ -595,7 +376,7 @@ class TestBackpressureHandler: mock_exchange ) # Make it return the mock exchange await handler.publish_backpressure_request( - ScaleRequestV1("test_service", "test_task", 100, 50, ScalingRequestAlert.UPDATE) + ScaleRequestV1("test_service", "test_task", 50) ) assert mock_async_run.call_count == 2 diff --git a/tests/model/test_model.py b/tests/model/test_model.py index b636c6a..a5b0761 100644 --- a/tests/model/test_model.py +++ b/tests/model/test_model.py @@ -9,7 +9,6 @@ from amqp.model.model import ( DataMessage, DataResponse, ScalingRequest, - ScalingRequestAlert, ) from amqp.model.snowflake_id import SnowflakeId @@ -159,9 +158,7 @@ class TestScalingRequest(unittest.TestCase): self.request = ScalingRequest( service_id="test_service", task_id="test_task", - max_availability=5, current_availability=3, - request_type=ScalingRequestAlert.OVERLOAD, version=1, ) @@ -171,9 +168,7 @@ class TestScalingRequest(unittest.TestCase): self.assertEqual(data["serviceId"], "test_service") self.assertEqual(data["taskId"], "test_task") - self.assertEqual(data["maxAvailability"], 5) self.assertEqual(data["currentAvailability"], 3) - self.assertEqual(data["requestType"], "OVERLOAD") self.assertEqual(data["version"], 1) def test_from_json(self): @@ -181,9 +176,7 @@ class TestScalingRequest(unittest.TestCase): { "serviceId": "test_service", "taskId": "test_task", - "maxAvailability": 5, "currentAvailability": 3, - "requestType": "OVERLOAD", "version": 1, } ) @@ -192,9 +185,7 @@ class TestScalingRequest(unittest.TestCase): self.assertEqual(request.service_id, "test_service") self.assertEqual(request.task_id, "test_task") - self.assertEqual(request.max_availability, 5) self.assertEqual(request.current_availability, 3) - self.assertEqual(request.request_type, ScalingRequestAlert.OVERLOAD) self.assertEqual(request.version, 1) diff --git a/tests/service/test_amq_service.py b/tests/service/test_amq_service.py index 0c57e9c..5e0d8a9 100644 --- a/tests/service/test_amq_service.py +++ b/tests/service/test_amq_service.py @@ -1,8 +1,7 @@ import unittest -from unittest.mock import AsyncMock, MagicMock, Mock, patch +from unittest.mock import AsyncMock, Mock, patch from amqp.config.amq_configuration import AMQAdapter, AMQConfiguration -from amqp.model.model import DataMessage from amqp.service.amq_service import AMQService @@ -146,21 +145,21 @@ class TestAMQService(unittest.IsolatedAsyncioTestCase): # Verify that client was reinitialized mock_client.init.assert_called_once() - def test_send_message(self): - # Create mock message and future - mock_message = Mock(spec=DataMessage) - mock_message.id.return_value = "test-id" - mock_future = MagicMock() - - # Setup mock message handler - self.service.amq_data_message_handler = Mock() - self.service.amq_data_message_handler.outstanding = {} - - # Send message - self.service.send_message(mock_message, mock_future) - - # Verify message was added to outstanding - self.assertEqual(self.service.amq_data_message_handler.outstanding["test-id"], mock_future) + # def test_send_message(self): + # # Create mock message and future + # mock_message = Mock(spec=DataMessage) + # mock_message.id.return_value = "test-id" + # mock_future = MagicMock() + # + # # Setup mock message handler + # self.service.amq_data_message_handler = Mock() + # self.service.amq_data_message_handler.outstanding = {} + # + # # Send message + # self.service.send_message(mock_message, mock_future) + # + # # Verify message was added to outstanding + # self.assertEqual(self.service.amq_data_message_handler.outstanding["test-id"], mock_future) def test_remove_outstanding_future(self): # Setup mock message handler with outstanding message