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

Merge pull request #4 from Aignosi/SIENTIAPDE-1049-lidar-com-erros-de-reconexao-no-ingestor-opc

Browse Source 
Sientiapde 1049 lidar com erros de reconexao no ingestor opc
This commit is contained in:
Matheus Demoner
2025-05-21 13:09:36 -03:00
committed by GitHub
parent 272545bff8 26356c05dd
commit 8b2e158f72
16 changed files with 321 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.coveragerc Normal file
View File

@@ -0,0 +1,3 @@
[run]
omit =
ingestor/app.py

3
.github/workflows/quality-gate.yml vendored
View File

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

2
README.md
View File

@@ -1,4 +1,4 @@
# sientia-dataops-opc-gateway
# sientia-dataops-opc-ingestor
OPC gateway to manage Scouter pipelines
## Local tests

0
app.py
View File

26
docker-compose.yaml
View File

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

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()
# Sleep for poll interval
sleep(ingestor.poll_interval)
exit_signal.wait(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()

103
ingestor/ingestor.py
View File

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

15
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:
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."""

106
ingestor/managers/ingestor_manager.py
View File

@@ -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,19 +172,12 @@ class IngestorManager():
if is_lost:
self.logger.warning(
f"OPC server {server} is lost. "
f"Desconnecting from server."
)
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']
f"Server will be disconnected."
)
for slot, _config in self.managed_tags.items():
self.managed_tags[slot].pop(server, None)
def declare_active(self):
"""
Declares the ingestor as active by sending a heartbeat signal to the resource manager.
@@ -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})
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:
"""

31
ingestor/managers/opc_manager.py
View File

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

16
simulator/redis-feeder.py
View File

@@ -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': {

30
tests/unit/managers/test_data_manager.py
View File

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

51
tests/unit/managers/test_ingestor_manager.py
View File

@@ -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"}
)

17
tests/unit/managers/test_opc_manager.py
View File

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

52
tests/unit/test_ingestor.py
View File

@@ -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")
]
)

19
values.yaml
View File

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