diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..2249368 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,3 @@ +[run] +omit = + ingestor/app.py diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 7efa623..e13ecb2 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -80,4 +80,5 @@ jobs: -Dsonar.host.url=$SONAR_HOST_URL \ -Dsonar.token=$SONAR_TOKEN \ -Dsonar.python.version=3.11 \ - -Dsonar.projectVersion=1.0.0 + -Dsonar.projectVersion=1.0.1 \ + -Dsonar.coverage.exclusions=ingestor/app.py diff --git a/README.md b/README.md index 0a90636..415658b 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# sientia-dataops-opc-gateway +# sientia-dataops-opc-ingestor OPC gateway to manage Scouter pipelines ## Local tests diff --git a/app.py b/app.py deleted file mode 100644 index e69de29..0000000 diff --git a/docker-compose.yaml b/docker-compose.yaml index 891fc9b..2216ee5 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -1,19 +1,19 @@ version: '3.8' services: - ingestor: - build: - context: . - environment: - HOSTNAME: ingestor - container_name: ingestor - depends_on: - - kafka - - redis - networks: - - kafka-net - env_file: - - .env + # ingestor: + # build: + # context: . + # environment: + # HOSTNAME: ingestor + # container_name: ingestor + # depends_on: + # - kafka + # - redis + # networks: + # - kafka-net + # env_file: + # - .env zookeeper: image: confluentinc/cp-zookeeper:latest diff --git a/ingestor/app.py b/ingestor/app.py index b9885fc..7b3f05b 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -1,19 +1,42 @@ -from time import sleep +from threading import Event +import signal +import os +import traceback from ingestor.ingestor import Ingestor +exit_signal = Event() + def main(): ingestor = Ingestor() - ingestor.prepare_ingestor() + ingestor.logger.info("Ingestor prepared. Starting main loop.") - while True: + while not exit_signal.is_set(): + try: + ingestor.loop() - ingestor.loop() + exit_signal.wait(ingestor.poll_interval) - # Sleep for poll interval - sleep(ingestor.poll_interval) + except Exception: + print("Exception in main loop. Setting exit_signal flag.") + traceback.print_exc() + exit_signal.set() + + ingestor.shutdown() + + ingestor.logger.info("Main loop exit_signaled.") + + os._exit(0) + + +def signal_handler(_signum, _frame): + print(f"Received signal {_signum}. Setting exit_signal flag.") + exit_signal.set() if __name__ == "__main__": + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + signal.signal(signal.SIGHUP, signal_handler) main() diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 3d39b35..b508b56 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -1,8 +1,11 @@ from logging import Formatter, StreamHandler, getLogger from os import getenv +from copy import deepcopy +from typing import Dict, Any + +from sientia_do.notifications.handlers import NotificationHandler from ingestor.managers.ingestor_manager import IngestorManager -from sientia_do.notifications.handlers import NotificationHandler class Ingestor: @@ -10,7 +13,8 @@ class Ingestor: """ Initializes the ingestor with configuration values retrieved from environment variables. Environment Variables: - KAFKA_SERVERS (str): Comma-separated list of Kafka server addresses. Defaults to "localhost:9092". + KAFKA_SERVERS (str): Comma-separated list of Kafka server addresses. + Defaults to "localhost:9092". REDIS_HOST (str): Hostname of the Redis server. Defaults to "localhost". REDIS_PORT (int): Port number of the Redis server. Defaults to 6379. LEASE_TTL (int): Time-to-live for leases in seconds. Defaults to 10. @@ -29,13 +33,13 @@ class Ingestor: kafka_servers = getenv("KAFKA_SERVERS", "localhost:9092") self.redis_host = getenv("REDIS_HOST", "localhost") - self.redis_port = int(getenv("REDIS_PORT", 6379)) + self.redis_port = int(getenv("REDIS_PORT", '6379')) self.redis_username = getenv("REDIS_USERNAME", None) self.redis_password = getenv("REDIS_PASSWORD", None) - self.lease_ttl = int(getenv("LEASE_TTL", 10)) - self.heartbeat_ttl = int(getenv("HEARTBEAT_TTL", 20)) + self.lease_ttl = int(getenv("LEASE_TTL", '10')) + self.heartbeat_ttl = int(getenv("HEARTBEAT_TTL", '20')) self.pod_id = getenv("HOSTNAME", "localhost") - self.poll_interval = int(getenv("POLL_INTERVAL", 5)) + self.poll_interval = int(getenv("POLL_INTERVAL", '5')) self.kafka_servers = kafka_servers.split(",") self.logger = None @@ -49,9 +53,15 @@ class Ingestor: model_name="-", model="-" ) - # build args for build notificarions components - # call build notifications components + self.ingestor_manager = None + + def shutdown(self): + if self.ingestor_manager: + self.ingestor_manager.shutdown() + + def __del__(self): + self.shutdown() def init_logger(self): """ @@ -128,7 +138,7 @@ class Ingestor: # Get slot lease acquired = self.ingestor_manager.get_slot_leases() - self.logger.info(f"Acquired slots: {acquired}") + self.logger.info("Acquired slots: %s", acquired) self.handle_acquired_tags(acquired) @@ -147,9 +157,7 @@ class Ingestor: # This ingestor is active and has no slots, so we need to try to # Get slot lease - acquired = self.ingestor_manager.get_slot_leases(1) - - self.handle_acquired_tags(acquired) + self.ingestor_manager.get_slot_leases(1) def manage_leases(self, available_slots: int, lacking_ingestors: int, slot_diff: int): """ @@ -171,16 +179,14 @@ class Ingestor: 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}") + self.logger.info("Slots available: %s", available_slots) # Get slot lease - acquired = self.ingestor_manager.get_slot_leases(available_slots) + self.ingestor_manager.get_slot_leases(available_slots) - self.handle_acquired_tags(acquired) + elif lacking_ingestors <= 0 and slot_diff > 0: - elif lacking_ingestors == 0 and slot_diff > 0: - - self.logger.info(f"Extra slots available: {slot_diff}") + self.logger.info("Extra slots available: %s", 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 @@ -188,6 +194,40 @@ class Ingestor: self.ingestor_manager.drop_slot_leases(overleases) + for lease in overleases: + self.ingestor_manager.unsubscribe_slot(lease) + self.ingestor_manager.managed_tags.pop(lease) + + def update_ingestor_manager(self, old_managed_tags: Dict[str, Any]): + """ + Updates the ingestor manager with the new managed tags. + Args: + old_managed_tags (Dict[str, Any]): The old managed tags. + """ + + self.logger.debug("Current managed tags: %s", + self.ingestor_manager.managed_tags) + + self.ingestor_manager.update_opc_servers() + new_managed_tags = deepcopy(self.ingestor_manager.managed_tags) + + self.logger.debug("Comparing new managed tags %s" + "with old managed tags %s", + new_managed_tags, old_managed_tags) + + for slot, config in new_managed_tags.items(): + if slot not in old_managed_tags: + self.ingestor_manager.subscribe_to_tags({slot: config}) + continue + + if config != old_managed_tags[slot]: + self.ingestor_manager.unsubscribe_slot(slot) + self.ingestor_manager.subscribe_to_tags({slot: config}) + + for slot in old_managed_tags.keys(): + if slot not in new_managed_tags: + self.ingestor_manager.unsubscribe_slot(slot) + def loop(self): """ Executes the main loop for managing ingestors and slots. @@ -211,33 +251,50 @@ class Ingestor: self.logger.info("Polling for slot updates...") # Get active ingestors + + current_managed_tags = deepcopy(self.ingestor_manager.managed_tags) + 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.logger.debug("Managing no slots...") self.manage_no_slots(number_of_slots) 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.logger.debug("Managing leases...") self.manage_leases(available_slots, lacking_ingestors, slot_diff) self.logger.debug( - 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}" + "Active ingestors: %s, " + "Number of slots: %s, " + "Number of leases: %s, " + "Managed tags: %s, " + "Managed servers: %s", + ingestors, + number_of_slots, + number_of_leases, + self.ingestor_manager.managed_tags, + self.ingestor_manager.opc_managers ) if not self.ingestor_manager.managed_tags: # No slots acquired self.logger.info("No slots acquired in this loop") # Update opc servers + self.logger.debug("Updating slot config...") self.ingestor_manager.update_slot_config() # Check OPC cycles + self.logger.debug("Checking OPC servers integrity...") self.ingestor_manager.check_opc_servers_integrity() + + self.logger.debug("Updating managed tags...") + self.update_ingestor_manager( + current_managed_tags + ) diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 3f9ca3c..b16b933 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -51,14 +51,21 @@ class DataManager(): self.logger = logger self.notification_handler = notification_handler - def __del__(self): - """Destructor to close the producer connection.""" - print("Closing Kafka producer...") + def shutdown(self): + """Closes the Kafka producer connection.""" if self.kafka_producer: - self.kafka_producer.flush(timeout=10) - self.kafka_producer.close() + try: + self.kafka_producer.flush(timeout=10) + self.kafka_producer.close() + except Exception as e: + self.logger.error( + f"Error closing Kafka producer: {e}") else: - print("Kafka producer is already closed or not initialized.") + self.logger.warning( + "Kafka producer is already closed or not initialized.") + + def __del__(self): + self.shutdown() def delivery_report(self, msg: str): """Callback for delivery reports from Kafka.""" diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 8c05c44..433fcef 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -2,11 +2,11 @@ from logging import Logger import traceback from typing import Dict, List from copy import deepcopy +from sientia_do.notifications.handlers import NotificationHandler +from sientia_do.notifications.models import NotificationLevel from ingestor.managers.data_manager import DataManager from ingestor.managers.opc_manager import OpcManager from ingestor.managers.resource_manager import ResourceManager -from sientia_do.notifications.models import Notification, NotificationLevel -from sientia_do.notifications.handlers import NotificationHandler class IngestorManager(): @@ -30,7 +30,8 @@ class IngestorManager(): self.notification_handler = notification_handler - def initialize_opc_from_config(self, server_config: dict, data_manager: DataManager, logger: Logger) -> OpcManager | None: + def initialize_opc_from_config(self, server_config: dict, + data_manager: DataManager, logger: Logger) -> OpcManager | None: """ Initializes an OPC Manager instance using the provided server configuration. Args: @@ -53,9 +54,11 @@ class IngestorManager(): self.logger.info( f"Initializing OpcManager at {server_config['url']}") manager = OpcManager( - server_config['name'], server_config['url'], data_manager, logger, server_config['server_uri'], - self.notification_handler, server_config.get('cert_path'), server_config.get( - 'private_key_path'), server_config.get('server_cert_path') + server_config['name'], server_config['url'], + data_manager, logger, server_config['server_uri'], + self.notification_handler, server_config.get('cert_path'), + server_config.get('private_key_path'), + server_config.get('server_cert_path') ) manager.config = server_config @@ -76,6 +79,14 @@ class IngestorManager(): return manager + def shutdown(self): + for _server_name, server in self.opc_managers.items(): + server.disconnect() + self.data_manager.shutdown() + + def __del__(self): + self.shutdown() + def update_opc_servers(self): """ Updates the OPC (OLE for Process Control) server connections managed by the ingestor. @@ -91,7 +102,8 @@ class IngestorManager(): 4. Disconnects and removes OPC servers that are no longer registered. Attributes: self.managed_tags (dict): A nested dictionary containing slot and server configurations. - self.opc_managers (dict): A dictionary mapping server names to their OPC manager instances. + self.opc_managers (dict): A dictionary mapping server names to their + OPC manager instances. self.data_manager: An object responsible for managing data operations. self.logger: A logging object for recording warnings and other messages. Raises: @@ -101,27 +113,44 @@ class IngestorManager(): """ registered_servers = [] - for slot, slot_config in self.managed_tags.items(): + current_managed_tags = deepcopy(self.managed_tags) + for _slot, slot_config in current_managed_tags.items(): for server, server_config in slot_config.items(): registered_servers.append(server) - server_config = server_config.copy() + server_config = deepcopy(server_config) server_config.pop('tags', None) - server_instance = None - if server not in self.opc_managers: - + server_instance = self.opc_managers.get(server, None) + if server_instance is None: + self.logger.debug( + f"Initializing OPC manager for server {server}" + ) server_instance = self.initialize_opc_from_config( server_config, self.data_manager, self.logger ) - elif self.opc_managers[server].config != server_config: - self.opc_managers[server].disconnect() + elif server_instance.config != server_config: + self.logger.warning( + f"Reinitializing OPC manager for server {server}" + ) + server_instance.disconnect() del self.opc_managers[server] server_instance = self.initialize_opc_from_config( server_config, self.data_manager, self.logger ) + else: + self.logger.debug( + f"OPC manager for server {server} is already initialized and up to date" + ) if server_instance is not None: self.opc_managers[server] = server_instance + else: + self.logger.warning( + f"Failed to initialize OPC manager for server {server}, " + f"removing server from managed tags." + ) + _a = [self.managed_tags[slot].pop(server, None) + for slot, _value in current_managed_tags.items()] for server in list(self.opc_managers.keys()): if server not in registered_servers: @@ -143,18 +172,11 @@ class IngestorManager(): if is_lost: self.logger.warning( f"OPC server {server} is lost. " - f"Desconnecting from server." + f"Server will be disconnected." ) - opc_manager.disconnect() - self.opc_managers[server] = self.initialize_opc_from_config( - opc_manager.config, self.data_manager, self.logger - ) - self.update_opc_servers() - for slot, slot_config in self.managed_tags.items(): - if server in slot_config: - self.manage_server( - slot, server, slot_config[server], slot_config[server]['tags'] - ) + + for slot, _config in self.managed_tags.items(): + self.managed_tags[slot].pop(server, None) def declare_active(self): """ @@ -212,14 +234,16 @@ class IngestorManager(): Dict: A dictionary where the keys are the slot identifiers (as strings) and the values are the leased slot details. Behavior: - - Iterates through available slots and attempts to lease them using the resource manager. + - Iterates through available slots and attempts to lease + them using the resource manager. - Logs the leasing of each slot. - Updates the `managed_tags` attribute with the acquired slots. - Stops leasing once the specified `max_slots` are acquired. - - If unable to acquire the requested number of slots, logs a warning and returns the slots that were leased. + - If unable to acquire the requested number of slots, + logs a warning and returns the slots that were leased. Notes: - - If a slot is leased but its details cannot be retrieved (i.e., `get_tag_slot` returns None), - that slot is skipped. + - If a slot is leased but its details cannot be retrieved + (i.e., `get_tag_slot` returns None), that slot is skipped. """ acquired = {} @@ -280,33 +304,17 @@ class IngestorManager(): removed_slots = [] update = {} - for slot, slot_config in self.managed_tags.items(): + for slot, _slot_config in self.managed_tags.items(): self.resource_manager.renew_tag_lease(slot) update = self.resource_manager.get_tag_slot(slot) if update is None: - self.logger.warning( - f"Slot {slot} configuration not found. " - f"Removing slot from managed tags." - ) - self.unsubscribe_slot(slot) removed_slots.append(slot) - continue - if update != slot_config: - self.logger.info( - f"Slot {slot} configuration updated. " - f"Old: {slot_config}, New: {update}" - ) - self.managed_tags[slot] = update - - self.unsubscribe_slot(slot) - self.update_opc_servers() - self.subscribe_to_tags({slot: update}) + self.managed_tags[slot] = update for slot in removed_slots: - del self.managed_tags[slot] - self.update_opc_servers() + self.managed_tags.pop(slot, None) def drop_slot_leases(self, ids: List[str]) -> None: """ @@ -321,8 +329,8 @@ class IngestorManager(): None """ - for id in ids: - self.resource_manager.drop_tag_lease(id) + for lease_id in ids: + self.resource_manager.drop_tag_lease(lease_id) def manage_server(self, slot: str, server: str, server_config: dict, tags: dict) -> int: """ diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index b8065f3..e218956 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -188,13 +188,21 @@ class OpcManager(): self.logger.warning("Client already disconnected.") return try: - [self.subscriptions[sub].delete() for sub in self.subscriptions] + _a = [self.subscriptions[sub].delete() + for sub in self.subscriptions] self.logger.warning("Deleted all subscriptions.") + except Exception as sub_error: + self.logger.error(f"Failed to clean up subscription: {sub_error}") + + try: + self.client.disconnect() + except Exception as conn_error: + self.logger.error( + f"Failed to disconnect from OPC UA server: {conn_error}") + finally: del self.client self.client = None self.logger.warning("Disconnected from OPC UA server.") - except Exception as sub_error: - self.logger.error(f"Failed to clean up subscription: {sub_error}") def datachange_notification(self, node, _val, data): """ @@ -223,10 +231,8 @@ class OpcManager(): tag = str(node) self.logger.debug( - f"Data change notification received for tag: {tag} after {self.nodes[tag]['cycle_rule']['cycle_count']} cycles") - - self.logger.debug( - f"Resetting cycle count for tag: {tag} after {self.non_receive_count} OPC cycles") + f"Data change notification received for tag:" + f"{tag} after {self.nodes[tag]['cycle_rule']['cycle_count']} cycles") self.nodes[tag]['cycle_rule']['cycle_count'] = 0 self.non_receive_count = 0 @@ -238,12 +244,13 @@ class OpcManager(): 'value': value } - [self.data_manager.publish(e, data) - for e in self.nodes[tag]['topics']] + _a = [self.data_manager.publish(e, data) + for e in self.nodes[tag]['topics']] def check_cycles(self): """ - Checks the cycle counts for all monitored nodes and sends notifications if thresholds are exceeded. + Checks the cycle counts for all monitored nodes and sends + notifications if thresholds are exceeded. This method iterates through all monitored nodes and updates their cycle counts based on configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers @@ -251,7 +258,8 @@ class OpcManager(): """ for node, config in self.nodes.items(): - self.nodes[node]['cycle_rule']['cycle_count'] += self.nodes[node]['cycle_rule']['cycle_increment'] + self.nodes[node]['cycle_rule']['cycle_count'] += config[ + 'cycle_rule']['cycle_increment'] if self.nodes[node]['cycle_rule']['cycle_count'] >= 5: name = config['tag_name'] cycles = self.nodes[node]['cycle_rule']['cycle_count'] @@ -273,7 +281,8 @@ class OpcManager(): if self.non_receive_count >= 5: self.notification_handler.build_and_send_notification( notification_id=f'OPC_LISTENNING_STOPPED__{self.name}', - message=f'{self.non_receive_count} cycles without receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}', + message=f'{self.non_receive_count} cycles without ' + f'receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}', block="opc_manager", level=NotificationLevel.ERROR ) diff --git a/simulator/redis-feeder.py b/simulator/redis-feeder.py index 3f7e36a..d3dde93 100644 --- a/simulator/redis-feeder.py +++ b/simulator/redis-feeder.py @@ -1,14 +1,18 @@ import redis import json -import os # Redis connection settings -redis_host = "localhost" -redis_port = 6379 +REDIS_HOST = "localhost" +REDIS_PORT = 6379 +REDIS_USERNAME = None # "default" +REDIS_PASSWORD = None # "bdnZOpcyiL" + +OPC_URL = "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4840" +OPC_URL = "opc.tcp://localhost:4840" # Connect to Redis -r = redis.Redis(host=redis_host, port=redis_port, - decode_responses=True, username='default', password='bdnZOpcyiL') +r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, + decode_responses=True, username=REDIS_USERNAME, password=REDIS_PASSWORD) # Define the key pattern to target pattern = "slot:opc_tags:*" @@ -25,7 +29,7 @@ new_data = { "slot:opc_tags:1": { "server1": { "name": "server1", - "url": "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4840", + "url": OPC_URL, "server_uri": "http://opcua-server.simulator", "tags": { 'ns=2;i=2': { diff --git a/tests/unit/managers/test_data_manager.py b/tests/unit/managers/test_data_manager.py index 34c387a..80f75e0 100644 --- a/tests/unit/managers/test_data_manager.py +++ b/tests/unit/managers/test_data_manager.py @@ -108,31 +108,45 @@ def test___init___failure_max_attempts(kafka): assert False, "Expected NoBrokersAvailable exception was not raised." -def test___del___has_producer(data_manager): +def test_shutdown_has_producer(data_manager): flush_mock = MagicMock() close_mock = MagicMock() data_manager.kafka_producer.flush = flush_mock data_manager.kafka_producer.close = close_mock - data_manager.__del__() + data_manager.shutdown() flush_mock.assert_called_once() close_mock.assert_called_once() -@patch("ingestor.managers.data_manager.print") -def test___del___no_producer(print, data_manager): +def test_shutdown_no_producer(data_manager): data_manager.kafka_producer = None - # Call the __del__ method - data_manager.__del__() + data_manager.shutdown() - # Check if the print statement was called - print.assert_any_call( + data_manager.logger.warning.assert_any_call( "Kafka producer is already closed or not initialized." ) +def test_shutdown_exception(data_manager): + data_manager.kafka_producer.flush = MagicMock( + side_effect=Exception("Test error")) + data_manager.kafka_producer.close = MagicMock() + + data_manager.shutdown() + data_manager.logger.error.assert_called_once_with( + "Error closing Kafka producer: Test error" + ) + + +def test___del__(data_manager): + data_manager.shutdown = MagicMock() + data_manager.__del__() + data_manager.shutdown.assert_called_once() + + def test_delivery_report(data_manager): msg = MagicMock() msg.topic = "test_topic" diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 6e59103..fa3f7be 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -331,14 +331,6 @@ def test_update_slot_config(ingestor_manager): "config": "new_config"} assert "slot3" not in ingestor_manager.managed_tags - assert ingestor_manager.update_opc_servers.call_count == 2 - ingestor_manager.subscribe_to_tags.assert_called_once_with( - {'slot1': {"config": "updated_config"}} - ) - ingestor_manager.unsubscribe_slot.assert_any_call("slot3") - ingestor_manager.unsubscribe_slot.assert_any_call("slot1") - assert ingestor_manager.unsubscribe_slot.call_count == 2 - ingestor_manager.resource_manager.renew_tag_lease.assert_any_call("slot1") ingestor_manager.resource_manager.renew_tag_lease.assert_any_call("slot3") ingestor_manager.resource_manager.renew_tag_lease.assert_any_call("slot2") @@ -546,27 +538,9 @@ def test_check_opc_servers_integrity_server_lost(ingestor_manager): ingestor_manager.initialize_opc_from_config = MagicMock( return_value=new_manager) - # Mock update_opc_servers and manage_server - ingestor_manager.update_opc_servers = MagicMock() - ingestor_manager.manage_server = MagicMock() - # Call the method ingestor_manager.check_opc_servers_integrity() - # Verify that the lost server was disconnected - opc_manager.disconnect.assert_called_once() - - # Verify that a new manager was initialized - ingestor_manager.initialize_opc_from_config.assert_called_once_with( - opc_manager.config, ingestor_manager.data_manager, ingestor_manager.logger - ) - - # Verify that the new manager was assigned - assert ingestor_manager.opc_managers["server1"] == new_manager - - # Verify that update_opc_servers was called - ingestor_manager.update_opc_servers.assert_called_once() - def test_check_opc_servers_integrity_server_lost_with_tags(ingestor_manager): # Setup mock OPC manager that will be lost @@ -594,30 +568,5 @@ def test_check_opc_servers_integrity_server_lost_with_tags(ingestor_manager): ingestor_manager.initialize_opc_from_config = MagicMock( return_value=new_manager) - # Mock update_opc_servers and manage_server - ingestor_manager.update_opc_servers = MagicMock() - ingestor_manager.manage_server = MagicMock() - # Call the method ingestor_manager.check_opc_servers_integrity() - - # Verify that the lost server was disconnected - opc_manager.disconnect.assert_called_once() - - # Verify that a new manager was initialized - ingestor_manager.initialize_opc_from_config.assert_called_once_with( - opc_manager.config, ingestor_manager.data_manager, ingestor_manager.logger - ) - - # Verify that the new manager was assigned - assert ingestor_manager.opc_managers["server1"] == new_manager - - # Verify that update_opc_servers was called - ingestor_manager.update_opc_servers.assert_called_once() - - # Verify that manage_server was called with the correct tags - ingestor_manager.manage_server.assert_called_once_with( - "slot1", "server1", - {"config": "config1", "tags": {"tag1": "value1"}}, - {"tag1": "value1"} - ) diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index 8886709..a57ef46 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -188,7 +188,7 @@ def test_disconnect_success(opc_manager_subscribed): assert opc_manager_subscribed.client is None -def test_disconnect_error(opc_manager_subscribed): +def test_disconnect_error_unsubscribe(opc_manager_subscribed): opc_manager_subscribed.client = MagicMock() opc_manager_subscribed.subscriptions['sub1'] = MagicMock( delete=MagicMock(side_effect=Exception("Test error")) @@ -197,11 +197,24 @@ def test_disconnect_error(opc_manager_subscribed): opc_manager_subscribed.disconnect() opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() - opc_manager_subscribed.client.disconnect.assert_not_called() + opc_manager_subscribed.client = None opc_manager_subscribed.logger.error.assert_called_once_with( "Failed to clean up subscription: Test error") +def test_disconnect_error(opc_manager_subscribed): + opc_manager_subscribed.client = MagicMock() + opc_manager_subscribed.client.disconnect = MagicMock( + side_effect=Exception("Test error")) + + opc_manager_subscribed.disconnect() + + opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once() + opc_manager_subscribed.client = None + opc_manager_subscribed.logger.error.assert_called_once_with( + "Failed to disconnect from OPC UA server: Test error") + + def test_datachange_notification(opc_manager_subscribed): data = MagicMock( monitored_item=MagicMock( diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 448363c..54ebdc7 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -1,4 +1,5 @@ -from unittest.mock import ANY, MagicMock, patch +from unittest.mock import ANY, MagicMock, patch, call +from os import getenv from pytest import fixture from ingestor.ingestor import Ingestor @@ -24,13 +25,13 @@ def test___init__(notification_handler, init_logger, getenv): getenv.assert_any_call("KAFKA_SERVERS", "localhost:9092") getenv.assert_any_call("REDIS_HOST", "localhost") - getenv.assert_any_call("REDIS_PORT", 6379) + getenv.assert_any_call("REDIS_PORT", '6379') getenv.assert_any_call("REDIS_USERNAME", None) getenv.assert_any_call("REDIS_PASSWORD", None) - getenv.assert_any_call("LEASE_TTL", 10) - getenv.assert_any_call("HEARTBEAT_TTL", 20) + getenv.assert_any_call("LEASE_TTL", '10') + getenv.assert_any_call("HEARTBEAT_TTL", '20') getenv.assert_any_call("HOSTNAME", "localhost") - getenv.assert_any_call("POLL_INTERVAL", 5) + getenv.assert_any_call("POLL_INTERVAL", '5') assert ingestor.kafka_servers == ["localhost:9092", "localhost:35"] assert ingestor.redis_host == "localhost1" @@ -58,7 +59,7 @@ def test___init__(notification_handler, init_logger, getenv): @patch("ingestor.ingestor.getenv") @patch("ingestor.ingestor.Ingestor.init_logger") @patch("ingestor.ingestor.NotificationHandler") -def ingestor(notification_handler, init_logger, getenv): +def ingestor(_notification_handler, _init_logger, _getenv): ing = Ingestor() ing.logger = MagicMock() @@ -82,7 +83,8 @@ def test_init_logger(formatter, stream_handler, get_logger, ingestor): stream_handler.assert_called_once() formatter.assert_called_once_with( '%(asctime)s - %(name)s - %(levelname)s - %(message)s') - ingestor.logger.setLevel.assert_called_once_with("INFO") + ingestor.logger.setLevel.assert_called_once_with( + getenv("LOG_LEVEL", "INFO")) ingestor.logger.addHandler.assert_called_once_with( stream_handler.return_value) stream_handler.return_value.setFormatter.assert_called_once_with( @@ -172,8 +174,6 @@ def test_manage_slots_none_available_none_available(ingestor_manager_started): ingestor_manager_started.ingestor_manager.get_slot_leases.assert_called_once_with( 1) - ingestor_manager_started.handle_acquired_tags.assert_called_once_with( - ingestor_manager_started.ingestor_manager.get_slot_leases.return_value) def test_manage_leases_no_available_slots_no_extra_slots(ingestor_manager_started): @@ -193,8 +193,6 @@ def test_manage_leases_available_slots_innactive_ingestors(ingestor_manager_star 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() @@ -267,3 +265,35 @@ def test_loop_no_managed(ingestor_manager_started): ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once() ingestor_manager_started.logger.info.assert_any_call( "No slots acquired in this loop") + + +def test_update_ingestor_manager(ingestor_manager_started): + ingestor_manager_started.ingestor_manager.managed_tags = { + "slot_to_create": "new_config", + "slot_to_update": "new_config", + "slot_to_do_nothing": "old_config" + } + + old_managed_tags = { + "slot_to_update": "old_config", + "slot_to_delete": "old_config", + "slot_to_do_nothing": "old_config" + } + + ingestor_manager_started.update_ingestor_manager(old_managed_tags) + + ingestor_manager_started.ingestor_manager.update_opc_servers.assert_called_once() + ingestor_manager_started.ingestor_manager.subscribe_to_tags.assert_has_calls( + [ + call( + {"slot_to_create": "new_config"}), + call( + {"slot_to_update": "new_config"}) + ] + ) + ingestor_manager_started.ingestor_manager.unsubscribe_slot.assert_has_calls( + [ + call("slot_to_update"), + call("slot_to_delete") + ] + ) diff --git a/values.yaml b/values.yaml index 08d9a14..a94bca0 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.0.1" + tag: "0.0.2" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: @@ -19,7 +19,7 @@ imagePullSecrets: # This is to override the chart name. nameOverride: "sientia-opc-ingestor" fullnameOverride: "sientia-opc-ingestor" -namespace: sientia-opc +namespace: sientia # This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/ serviceAccount: @@ -123,7 +123,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-opc-ingestor.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-988-criar-ingestor-opc" + value: "SIENTIAPDE-1049-lidar-com-erros-de-reconexao-no-ingestor-opc" - name: PYTHON_APP value: "ingestor.app" @@ -150,13 +150,20 @@ env: value: "30" - name: POLL_INTERVAL value: "10" + - name: LOG_LEVEL + value: "DEBUG" ssh: enabled: true - secretName: git-ssh-key-temp + secretName: git-ssh-key-sientia-opc-ingestor sshPath: /mnt/.ssh knownHostsPath: /mnt/known_hosts -# kubectl create secret docker-registry docker-hub-secret --namespace sientia-opc --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp -# helm upgrade --install sientia-dataops-opc-ingestor sientia/sientia-module -n sientia-opc --create-namespace -f ./values.yaml \ No newline at end of file +# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp +# helm upgrade --install sientia-opc-ingestor sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.1.0-uat + +# kubectl create secret generic git-ssh-key-sientia-opc-ingestor \ +# --namespace sientia \ +# --from-file=ssh-privatekey=git_key \ +# --type=kubernetes.io/ssh-auth \ No newline at end of file