From da079c63201c225a7cb784af15068034fb13eea7 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 24 Apr 2025 08:42:56 -0300 Subject: [PATCH] SIENTIAPDE-988 Refactor ingestor lease management and update Docker configurations - Updated CMD in Dockerfile to use the correct application entry point. - Added docker compose command to remove volumes in README. - Removed unnecessary restart policy from ingestor service in docker-compose.yaml. - Refactored manage_leases method in Ingestor class to improve clarity and parameter naming. - Added get_number_of_leases method in IngestorManager to retrieve active leases. - Implemented get_all_leases method in ResourceManager to fetch active leases from Redis. - Enhanced unit tests for lease management in test_ingestor_manager and test_resource_manager. --- Dockerfile | 2 +- README.md | 1 + docker-compose.yaml | 1 - ingestor/ingestor.py | 51 +++++++++++--------- ingestor/managers/ingestor_manager.py | 13 +++++ ingestor/managers/resource_manager.py | 11 +++++ tests/unit/managers/test_ingestor_manager.py | 16 ++++++ tests/unit/managers/test_resource_manager.py | 9 ++++ tests/unit/test_ingestor.py | 14 +++--- 9 files changed, 87 insertions(+), 31 deletions(-) diff --git a/Dockerfile b/Dockerfile index b129853..2e48159 100644 --- a/Dockerfile +++ b/Dockerfile @@ -27,4 +27,4 @@ RUN --mount=type=ssh \ # Run the application -CMD ["python", "-m", "ingestor.ingestor"] \ No newline at end of file +CMD ["python", "-m", "ingestor.app"] \ No newline at end of file diff --git a/README.md b/README.md index 8f17c4c..4e54b12 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,7 @@ sudo systemctl restart docker ### Run docker compose ''' +docker compose down -v docker compose build --ssh default=$HOME/.ssh/id_ed25519_docker docker compose up -d ''' diff --git a/docker-compose.yaml b/docker-compose.yaml index ce7fed1..891fc9b 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -2,7 +2,6 @@ version: '3.8' services: ingestor: - restart: always build: context: . environment: diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 5e46c53..7568708 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -134,37 +134,38 @@ class Ingestor: self.handle_acquired_tags(acquired) - def manage_leases(self, ingestor_diff: int, slot_diff: int): + def manage_leases(self, available_slots: int, lacking_ingestors: int, slot_diff: int): """ - Manages the allocation and deallocation of slot leases based on the - differences in the number of active ingestors and available slots. + Manages the allocation and deallocation of slot leases for ingestors based on + the number of available slots, lacking ingestors, and slot differences. Args: - ingestor_diff (int): The difference between the required and available - ingestors. A positive value indicates that there are inactive - ingestors and available slots. - slot_diff (int): The difference between the required and available - slots. A positive value indicates that there are active ingestors - without assigned slots. + available_slots (int): The number of slots currently available for allocation. + lacking_ingestors (int): The number of ingestors that are active and without slots. + slot_diff (int): The difference between the total slots and the required slots. Behavior: - - If `ingestor_diff` is greater than 0, it means there are available - slots due to inactive ingestors. The method will acquire slot leases - for the available slots and handle the acquired tags. - - If `slot_diff` is greater than 0, it means there are active ingestors - without slots. The method will drop slot leases for the excess - managed tags. + - If there are available slots and lacking ingestors, attempts to acquire slot leases + for the available slots and processes the acquired tags. + - If there are no lacking ingestors but there are extra slots (slot_diff > 0), + releases the extra slot leases to ensure proper allocation. + Logs: + - Logs the number of available slots when attempting to acquire leases. + - Logs the number of extra slots when releasing leases. """ - if ingestor_diff > 0: - # Some ingestors are innactive, so theres "ingestor_diff" slots available - self.logger.info(f"Slots available: {ingestor_diff}") + if available_slots > 0 and lacking_ingestors > 0: + # Some ingestors are innactive, so theres "available_slots" slots available + self.logger.info(f"Slots available: {available_slots}") # Get slot lease - acquired = self.ingestor_manager.get_slot_leases(ingestor_diff) + acquired = self.ingestor_manager.get_slot_leases(available_slots) self.handle_acquired_tags(acquired) - elif slot_diff > 0: - # Some ingestors are active and without slots, so we need to drop + elif lacking_ingestors == 0 and slot_diff > 0: + + self.logger.info(f"Extra slots available: {slot_diff}") + # There's enough slots for all ingestors, but this ingestor has more than one slot + # So we need to drop the extra leases overleases = list(self.ingestor_manager.managed_tags.keys())[1:] @@ -194,19 +195,23 @@ class Ingestor: self.logger.info("Polling for slot updates...") # Get active ingestors ingestors = self.ingestor_manager.get_active_ingestors() + number_of_ingestors = len(ingestors) + number_of_leases = self.ingestor_manager.get_number_of_leases() number_of_slots = self.ingestor_manager.get_number_of_slots() # Handle no slots self.manage_no_slots(number_of_slots) - ingestor_diff = number_of_slots - len(ingestors) + available_slots = number_of_slots - number_of_leases + lacking_ingestors = number_of_slots - number_of_ingestors slot_diff = len(self.ingestor_manager.managed_tags) - 1 - self.manage_leases(ingestor_diff, slot_diff) + self.manage_leases(available_slots, lacking_ingestors, slot_diff) self.logger.info( f"Active ingestors: {ingestors}, " f"Number of slots: {number_of_slots}, " + f"Number of leases: {number_of_leases}, " f"Managed tags: {self.ingestor_manager.managed_tags}" f"Managed servers: {self.ingestor_manager.opc_managers}" ) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 25030cd..d5376b6 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -134,6 +134,19 @@ class IngestorManager(): ingestors = self.resource_manager.get_all_ingestors() return ingestors if ingestors else [] + def get_number_of_leases(self) -> int: + """ + Retrieves the number of leases managed by the resource manager. + This method fetches all available leases from the resource manager, + calculates their count, and updates the `number_of_slots` attribute. + Returns: + int: The total number of leases. Returns 0 if no leases are available. + """ + + leases = self.resource_manager.get_all_leases() + self.number_of_slots = len(leases) if leases else 0 + return self.number_of_slots + def get_number_of_slots(self) -> int: """ Retrieves the number of slots managed by the resource manager. diff --git a/ingestor/managers/resource_manager.py b/ingestor/managers/resource_manager.py index 2717f3f..275858a 100644 --- a/ingestor/managers/resource_manager.py +++ b/ingestor/managers/resource_manager.py @@ -117,3 +117,14 @@ class ResourceManager: """ return self.redis.keys("slot:opc_tags:*") + + def get_all_leases(self) -> List[str]: + """ + Retrieves all active leases from Redis. + This method fetches all keys in Redis that match the pattern for OPC tag leases + and returns a list of active leases. + Returns: + list: A list of active leases. + """ + + return self.redis.keys("lease:opc_tags:*") diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 5f718e5..991b375 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -183,6 +183,22 @@ def test_get_active_ingestors_empty(ingestor_manager): ingestor_manager.resource_manager.get_all_ingestors.assert_called_once() +def test_get_number_of_leases_success(ingestor_manager): + ingestor_manager.resource_manager.get_all_leases = MagicMock( + return_value=["lease1", "lease2"]) + result = ingestor_manager.get_number_of_leases() + assert result == 2 + ingestor_manager.resource_manager.get_all_leases.assert_called_once() + + +def test_get_number_of_leases_empty(ingestor_manager): + ingestor_manager.resource_manager.get_all_leases = MagicMock( + return_value=None) + result = ingestor_manager.get_number_of_leases() + assert result == 0 + ingestor_manager.resource_manager.get_all_leases.assert_called_once() + + def test_get_number_of_slots_success(ingestor_manager): ingestor_manager.resource_manager.get_all_slots = MagicMock( return_value=["slot1", "slot2"]) diff --git a/tests/unit/managers/test_resource_manager.py b/tests/unit/managers/test_resource_manager.py index c966562..aaa8fd2 100644 --- a/tests/unit/managers/test_resource_manager.py +++ b/tests/unit/managers/test_resource_manager.py @@ -98,3 +98,12 @@ def test_get_all_slots(resource_manager): resource_manager.redis.keys.assert_called_once_with( 'slot:opc_tags:*' ) + + +def test_get_all_leases(resource_manager): + resource_manager.redis.keys.return_value = ['lease1', 'lease2'] + result = resource_manager.get_all_leases() + assert result == ['lease1', 'lease2'] + resource_manager.redis.keys.assert_called_once_with( + 'lease:opc_tags:*' + ) diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 0ca5230..7e539a3 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -155,28 +155,30 @@ def test_manage_slots_none_available_none_available(ingestor_manager_started): ingestor_manager_started.ingestor_manager.get_slot_leases.return_value) -def test_manage_leases_0_0(ingestor_manager_started): +def test_manage_leases_no_available_slots_no_extra_slots(ingestor_manager_started): ingestor_manager_started.handle_acquired_tags = MagicMock() - ingestor_manager_started.manage_leases(0, 0) + ingestor_manager_started.manage_leases(0, 0, 0) ingestor_manager_started.ingestor_manager.get_slot_leases.assert_not_called() ingestor_manager_started.ingestor_manager.handle_acquired_tags.assert_not_called() ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_not_called() -def test_manage_leases_innactive_ingestors(ingestor_manager_started): +def test_manage_leases_available_slots_innactive_ingestors(ingestor_manager_started): ingestor_manager_started.handle_acquired_tags = MagicMock() - ingestor_manager_started.manage_leases(2, 0) + ingestor_manager_started.manage_leases(2, 2, 5) ingestor_manager_started.ingestor_manager.get_slot_leases.assert_called_once_with( 2) ingestor_manager_started.handle_acquired_tags.assert_called_once_with( ingestor_manager_started.ingestor_manager.get_slot_leases.return_value) + ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_not_called() -def test_manage_leases_available_ingestors(ingestor_manager_started): + +def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_started): ingestor_manager_started.handle_acquired_tags = MagicMock() ingestor_manager_started.ingestor_manager.managed_tags = { "tag1": "server1", @@ -184,7 +186,7 @@ def test_manage_leases_available_ingestors(ingestor_manager_started): "tag3": "server3" } - ingestor_manager_started.manage_leases(0, 2) + ingestor_manager_started.manage_leases(0, 0, 2) ingestor_manager_started.ingestor_manager.get_slot_leases.assert_not_called() ingestor_manager_started.handle_acquired_tags.assert_not_called()