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