gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1049

Browse Source 
Update ingestor and data manager for improved shutdown handling and logging

- Incremented project version in quality gate configuration.
- Commented out ingestor service in docker-compose for clarity.
- Enhanced main loop in app.py to handle exit signals and exceptions.
- Added shutdown methods in Ingestor and DataManager classes for graceful resource cleanup.
- Updated unit tests to validate shutdown behavior and exception handling.
- Introduced coverage configuration to omit specific files.
This commit is contained in:
vitor-aignosi
2025-05-20 17:19:41 -03:00
parent 92ed3c9cb2
commit c7dbe3585b
13 changed files with 235 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
ingestor/app.py
View File

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

57
ingestor/ingestor.py
View File

@@ -1,5 +1,7 @@
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
@@ -11,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.
@@ -53,6 +56,13 @@ class Ingestor:
self.ingestor_manager = None
def shutdown(self):
if self.ingestor_manager:
self.ingestor_manager.shutdown()
def __del__(self):
self.shutdown()
def init_logger(self):
"""
Initializes a logger instance for the class.
@@ -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):
"""
@@ -174,9 +182,7 @@ class Ingestor:
self.logger.info("Slots available: %s", available_slots)
# Get slot lease
acquired = self.ingestor_manager.get_slot_leases(available_slots)
self.handle_acquired_tags(acquired)
self.ingestor_manager.get_slot_leases(available_slots)
elif lacking_ingestors == 0 and slot_diff > 0:
@@ -188,6 +194,36 @@ class Ingestor:
self.ingestor_manager.drop_slot_leases(overleases)
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.
@@ -212,6 +248,8 @@ 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()
@@ -251,3 +289,8 @@ class Ingestor:
# 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
)

19
ingestor/managers/data_manager.py
View File

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

67
ingestor/managers/ingestor_manager.py
View File

@@ -79,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.
@@ -105,14 +113,14 @@ class IngestorManager():
"""
registered_servers = []
current_managed_tags = self.managed_tags.copy()
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}"
)
@@ -120,15 +128,19 @@ class IngestorManager():
server_config, self.data_manager, self.logger
)
elif self.opc_managers[server].config != server_config:
elif server_instance.config != server_config:
self.logger.warning(
f"Reinitializing OPC manager for server {server}"
)
self.opc_managers[server].disconnect()
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
@@ -153,29 +165,18 @@ class IngestorManager():
"""
Checks the integrity of the OPC servers and updates the OPC servers if necessary.
"""
current_managers = self.opc_managers.copy()
for server, opc_manager in current_managers.items():
for server, opc_manager in self.opc_managers.items():
opc_manager.check_cycles()
is_lost = opc_manager.check_opc_listenning()
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.logger.warning(
f"Reconnecting to OPC server {server}"
)
self.opc_managers.pop(server, None)
self.update_opc_servers()
if self.opc_managers.get(server) is not None:
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):
"""
@@ -303,35 +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)
self.logger.debug(
f"Comparing slot {slot} configuration: {slot_config} with update: {update}")
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:
"""

22
ingestor/managers/opc_manager.py
View File

@@ -188,7 +188,8 @@ 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}")
@@ -230,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
@@ -245,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
@@ -258,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']
@@ -280,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
)