.forgejo/workflows/build-and-publish.yml

@@ -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

.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

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:
 

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
         )
 
 

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

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.

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
 

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)
 
 

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